1use std::fmt;
2use std::fs::{self, OpenOptions};
3use std::io::{self, Write};
4use std::path::{Path, PathBuf};
5use std::sync::atomic::{AtomicBool, AtomicI64, AtomicU64, Ordering};
6use std::sync::Arc;
7use std::thread;
8use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
9
10use rand::Rng;
11use serde::{Deserialize, Serialize};
12
13use super::platform;
14
15pub const LOCK_FILE: &str = ".hayson.lock";
16
17const DEFAULT_STALE_MS: u64 = 30_000;
18const DEFAULT_HEARTBEAT_MS: u64 = 5_000;
19const DEFAULT_TIMEOUT_MS: u64 = 5_000;
20const DEFAULT_RETRY_MS: u64 = 50;
21const MAX_IMMEDIATE_RETRIES: u32 = 2;
22#[cfg(unix)]
24const HEARTBEAT_STEP_MS: u64 = 5;
25const TEMP_NAME_ATTEMPTS: u32 = 8;
26const OWNER_BYTES: usize = 16;
27
28fn now_ms() -> i64 {
29 SystemTime::now()
30 .duration_since(UNIX_EPOCH)
31 .map(|d| d.as_millis() as i64)
32 .unwrap_or(0)
33}
34
35fn mint_owner() -> String {
38 use std::fmt::Write as _;
39 let mut bytes = [0u8; OWNER_BYTES];
40 rand::rng().fill_bytes(&mut bytes);
41 let mut owner = String::with_capacity(OWNER_BYTES * 2);
42 for byte in bytes {
43 let _ = write!(owner, "{byte:02x}");
44 }
45 owner
46}
47
48#[derive(Debug, Clone, Copy, Default)]
60pub struct LockOptions {
61 pub stale_after_ms: Option<u64>,
62 pub heartbeat_ms: Option<u64>,
63 pub timeout_ms: Option<u64>,
64 pub retry_interval_ms: Option<u64>,
65 pub grace_served: Option<GraceServed>,
70}
71
72#[derive(Debug, Clone, Copy, PartialEq, Eq)]
81pub struct GraceServed {
82 owner: u128,
83 heartbeat_at: i64,
84}
85
86impl GraceServed {
87 pub fn of(holder: &LockHolder) -> Option<GraceServed> {
93 GraceServed::parse(&holder.owner, holder.heartbeat_at)
94 }
95
96 pub fn parse(owner: &str, heartbeat_at: i64) -> Option<GraceServed> {
103 if owner.len() != OWNER_BYTES * 2 {
104 return None;
105 }
106 u128::from_str_radix(owner, 16)
107 .ok()
108 .map(|owner| GraceServed { owner, heartbeat_at })
109 }
110
111 fn names(&self, holder: &LockHolder) -> bool {
112 GraceServed::of(holder) == Some(*self)
113 }
114}
115
116#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
119pub struct LockHolder {
120 pub owner: String,
121 #[serde(default)]
122 pub pid: Option<u32>,
123 #[serde(default)]
124 pub host: Option<String>,
125 pub started_at: i64,
126 pub heartbeat_at: i64,
127}
128
129fn new_holder(owner: &str) -> LockHolder {
130 let now = now_ms();
131 LockHolder {
132 owner: owner.to_string(),
133 pid: platform::pid(),
134 host: platform::hostname(),
135 started_at: now,
136 heartbeat_at: now,
137 }
138}
139
140#[derive(Debug, Clone)]
141pub struct LockedError {
142 pub subject: PathBuf,
143 pub holder: Option<LockHolder>,
144 pub waited_ms: u64,
145}
146
147impl fmt::Display for LockedError {
148 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
149 let by = match &self.holder {
150 Some(holder) => match (holder.pid, holder.host.as_deref()) {
151 (Some(pid), Some(host)) => format!("pid {pid} on {host}"),
152 (Some(pid), None) => format!("pid {pid}"),
153 (None, Some(host)) => format!("another writer on {host}"),
154 (None, None) => "another writer".to_string(),
155 },
156 None => "an unknown process".to_string(),
157 };
158 let waited = if self.waited_ms > 0 {
159 format!(" after waiting {} ms", self.waited_ms)
160 } else {
161 String::new()
162 };
163 write!(
164 f,
165 "database \"{}\" is locked for writing by {by}{waited} — raise LockOptions::timeout_ms to wait longer, or open it read-only",
166 self.subject.display(),
167 )
168 }
169}
170
171impl std::error::Error for LockedError {}
172
173#[derive(Debug)]
174pub enum DirLockError {
175 Locked(LockedError),
176 Lost(PathBuf),
179 SuspiciousLockFile(PathBuf),
182 Io(io::Error),
183}
184
185impl fmt::Display for DirLockError {
186 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
187 match self {
188 DirLockError::Locked(error) => error.fmt(f),
189 DirLockError::Lost(path) => write!(
190 f,
191 "the writer lock \"{}\" was taken over by another writer — this handle no longer holds it and must not write; reopen the database",
192 path.display(),
193 ),
194 DirLockError::SuspiciousLockFile(path) => write!(
195 f,
196 "the lock file \"{}\" is a symbolic link, not a regular file — refusing to read or replace it; remove whatever planted it",
197 path.display(),
198 ),
199 DirLockError::Io(error) => error.fmt(f),
200 }
201 }
202}
203
204impl std::error::Error for DirLockError {
205 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
206 match self {
207 DirLockError::Locked(error) => Some(error),
208 DirLockError::Io(error) => Some(error),
209 DirLockError::Lost(_) | DirLockError::SuspiciousLockFile(_) => None,
210 }
211 }
212}
213
214impl From<io::Error> for DirLockError {
215 fn from(error: io::Error) -> Self {
216 DirLockError::Io(error)
217 }
218}
219
220enum LockFileState {
221 Holder(LockHolder),
222 Missing,
223 Unreadable { touched_at_ms: i64 },
224}
225
226fn read_lock_file(path: &Path) -> Result<LockFileState, DirLockError> {
230 let metadata = match fs::symlink_metadata(path) {
231 Ok(metadata) => metadata,
232 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(LockFileState::Missing),
233 Err(_) => {
234 return Ok(LockFileState::Unreadable {
235 touched_at_ms: now_ms(),
236 })
237 }
238 };
239 if metadata.file_type().is_symlink() {
240 return Err(DirLockError::SuspiciousLockFile(path.to_path_buf()));
241 }
242 let touched_at_ms = metadata
243 .modified()
244 .ok()
245 .and_then(|time| time.duration_since(UNIX_EPOCH).ok())
246 .map(|d| d.as_millis() as i64)
247 .unwrap_or_else(now_ms);
248
249 let raw = match fs::read_to_string(path) {
250 Ok(raw) => raw,
251 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(LockFileState::Missing),
252 Err(_) => return Ok(LockFileState::Unreadable { touched_at_ms }),
253 };
254 match serde_json::from_str::<LockHolder>(&raw) {
255 Ok(holder) => Ok(LockFileState::Holder(holder)),
256 Err(_) => Ok(LockFileState::Unreadable { touched_at_ms }),
257 }
258}
259
260enum Abandonment {
261 Held,
262 LeaseExpired,
264 ProvablyDead,
267}
268
269fn provably_dead(holder: &LockHolder) -> bool {
270 let (Some(pid), Some(host)) = (holder.pid, holder.host.as_deref()) else {
271 return false;
272 };
273 if platform::hostname().as_deref() != Some(host) {
274 return false;
275 }
276 platform::process_alive(pid) == Some(false)
277}
278
279fn abandonment(holder: &LockHolder, stale_after_ms: u64) -> Abandonment {
280 if provably_dead(holder) {
281 return Abandonment::ProvablyDead;
282 }
283 if now_ms() - holder.heartbeat_at > stale_after_ms as i64 {
284 return Abandonment::LeaseExpired;
285 }
286 Abandonment::Held
287}
288
289fn temp_path_for(path: &Path, owner: &str, ordinal: u64) -> PathBuf {
290 PathBuf::from(format!("{}.{owner}.{ordinal}.tmp", path.display()))
291}
292
293fn write_temp(path: &Path, owner: &str, counter: &AtomicU64, holder: &LockHolder) -> io::Result<PathBuf> {
298 let json = serde_json::to_string(holder)?;
299 let mut refusal = None;
300 for _ in 0..TEMP_NAME_ATTEMPTS {
301 let temp = temp_path_for(path, owner, counter.fetch_add(1, Ordering::SeqCst));
302 let mut file = match OpenOptions::new().write(true).create_new(true).open(&temp) {
303 Ok(file) => file,
304 Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
305 refusal = Some(error);
306 continue;
307 }
308 Err(error) => return Err(error),
309 };
310 let written = file.write_all(json.as_bytes()).and_then(|()| file.sync_all());
311 drop(file);
312 if let Err(error) = written {
313 let _ = fs::remove_file(&temp);
314 return Err(error);
315 }
316 return Ok(temp);
317 }
318 Err(refusal.unwrap_or_else(|| {
319 io::Error::new(
320 io::ErrorKind::AlreadyExists,
321 "every candidate lock temporary was already taken",
322 )
323 }))
324}
325
326fn still_matches(path: &Path, observed: &LockHolder) -> Result<bool, DirLockError> {
327 Ok(match read_lock_file(path)? {
328 LockFileState::Holder(seen) => seen.owner == observed.owner && seen.heartbeat_at == observed.heartbeat_at,
329 LockFileState::Missing | LockFileState::Unreadable { .. } => false,
330 })
331}
332
333fn seize(
352 path: &Path,
353 observed: &LockHolder,
354 owner: &str,
355 counter: &AtomicU64,
356) -> Result<Option<LockHolder>, DirLockError> {
357 let holder = new_holder(owner);
358 let temp = write_temp(path, owner, counter, &holder)?;
359 match still_matches(path, observed) {
360 Ok(true) => {}
361 Ok(false) => {
362 let _ = fs::remove_file(&temp);
363 return Ok(None);
364 }
365 Err(error) => {
366 let _ = fs::remove_file(&temp);
367 return Err(error);
368 }
369 }
370 if let Err(error) = fs::rename(&temp, path) {
371 let _ = fs::remove_file(&temp);
372 return Err(error.into());
373 }
374 match read_lock_file(path)? {
375 LockFileState::Holder(landed) if landed.owner == holder.owner => Ok(Some(holder)),
376 _ => Ok(None),
377 }
378}
379
380fn finish_after_grace(
387 path: &Path,
388 observed: &LockHolder,
389 owner: &str,
390 counter: &AtomicU64,
391) -> Result<Option<LockHolder>, DirLockError> {
392 if !still_matches(path, observed)? {
393 return Ok(None);
394 }
395 seize(path, observed, owner, counter)
396}
397
398enum Pass {
401 Held(LockHolder),
403 Blocked(Option<LockHolder>),
406 GraceOwed(LockHolder),
409 Vacated,
412}
413
414fn one_pass(path: &Path, owner: &str, counter: &AtomicU64, stale_after_ms: u64) -> Result<Pass, DirLockError> {
415 if let Some(holder) = claim(path, owner)? {
416 return Ok(Pass::Held(holder));
417 }
418 match read_lock_file(path)? {
419 LockFileState::Holder(holder) => match abandonment(&holder, stale_after_ms) {
420 Abandonment::Held => Ok(Pass::Blocked(Some(holder))),
421 Abandonment::ProvablyDead => match seize(path, &holder, owner, counter)? {
422 Some(taken) => Ok(Pass::Held(taken)),
423 None => Ok(Pass::Blocked(Some(holder))),
424 },
425 Abandonment::LeaseExpired => Ok(Pass::GraceOwed(holder)),
426 },
427 LockFileState::Missing => Ok(Pass::Vacated),
428 LockFileState::Unreadable { touched_at_ms } => {
429 if now_ms() - touched_at_ms > stale_after_ms as i64 {
430 let _ = fs::remove_file(path);
431 Ok(Pass::Vacated)
432 } else {
433 Ok(Pass::Blocked(None))
434 }
435 }
436 }
437}
438
439fn claim(path: &Path, owner: &str) -> io::Result<Option<LockHolder>> {
440 let mut file = match OpenOptions::new().write(true).create_new(true).open(path) {
441 Ok(file) => file,
442 Err(error) if error.kind() == io::ErrorKind::AlreadyExists => return Ok(None),
443 Err(error) => return Err(error),
444 };
445 let holder = new_holder(owner);
446 let json = serde_json::to_string(&holder)?;
447 file.write_all(json.as_bytes())?;
448 file.sync_all()?;
449 Ok(Some(holder))
450}
451
452#[derive(Debug)]
455struct Beat {
456 path: PathBuf,
457 owner: String,
458 heartbeat_ms: u64,
459 counter: AtomicU64,
460 held: AtomicBool,
461 lost: AtomicBool,
462 stop: AtomicBool,
463 last_beat_ms: AtomicI64,
464}
465
466impl Beat {
467 fn lose(&self) -> DirLockError {
468 self.lost.store(true, Ordering::SeqCst);
469 DirLockError::Lost(self.path.clone())
470 }
471
472 fn renew(&self) -> Result<(), DirLockError> {
476 if self.lost.load(Ordering::SeqCst) {
477 return Err(DirLockError::Lost(self.path.clone()));
478 }
479 if !self.held.load(Ordering::SeqCst) {
480 return Ok(());
481 }
482 let holder = match read_lock_file(&self.path) {
483 Ok(LockFileState::Holder(holder)) if holder.owner == self.owner => holder,
484 Ok(_) => return Err(self.lose()),
485 Err(error) => {
486 self.lost.store(true, Ordering::SeqCst);
487 return Err(error);
488 }
489 };
490 let beaten = LockHolder {
491 heartbeat_at: now_ms(),
492 ..holder
493 };
494 let temp = write_temp(&self.path, &self.owner, &self.counter, &beaten)?;
495 if !self.held.load(Ordering::SeqCst) {
496 let _ = fs::remove_file(&temp);
497 return Ok(());
498 }
499 if let Err(error) = fs::rename(&temp, &self.path) {
500 let _ = fs::remove_file(&temp);
501 return Err(error.into());
502 }
503 self.last_beat_ms.store(now_ms(), Ordering::SeqCst);
504 Ok(())
505 }
506
507 fn renew_if_due(&self) -> Result<(), DirLockError> {
508 if self.lost.load(Ordering::SeqCst) {
509 return Err(DirLockError::Lost(self.path.clone()));
510 }
511 if !self.held.load(Ordering::SeqCst) {
512 return Ok(());
513 }
514 if now_ms() - self.last_beat_ms.load(Ordering::SeqCst) < self.heartbeat_ms as i64 {
515 return Ok(());
516 }
517 self.renew()
518 }
519}
520
521#[derive(Debug)]
534pub struct DirLock {
535 beat: Arc<Beat>,
536 #[cfg(unix)]
537 heartbeat: Option<thread::JoinHandle<()>>,
538}
539
540impl DirLock {
541 pub fn acquire(path: impl Into<PathBuf>, options: LockOptions) -> Result<DirLock, DirLockError> {
543 let path = path.into();
544 let subject = path.parent().map(Path::to_path_buf).unwrap_or_else(|| path.clone());
545 Self::acquire_as(path, options, subject)
546 }
547
548 pub fn acquire_as(path: PathBuf, options: LockOptions, subject: PathBuf) -> Result<DirLock, DirLockError> {
551 let stale_after_ms = options.stale_after_ms.unwrap_or(DEFAULT_STALE_MS);
552 let heartbeat_ms = options.heartbeat_ms.unwrap_or(DEFAULT_HEARTBEAT_MS);
553 let timeout_ms = options.timeout_ms.unwrap_or(DEFAULT_TIMEOUT_MS);
554 let retry_interval_ms = options.retry_interval_ms.unwrap_or(DEFAULT_RETRY_MS).max(1);
555 let grace = Duration::from_millis(stale_after_ms / 2);
556
557 let owner = mint_owner();
558 let counter = AtomicU64::new(0);
559 let started_at = Instant::now();
560 let deadline = started_at + Duration::from_millis(timeout_ms);
561 let mut retries = 0u32;
562 let mut blocker: Option<LockHolder> = None;
563 let mut last_seen: Option<LockHolder> = None;
564 let mut served = options.grace_served;
565
566 loop {
567 let mut retry_now = false;
568 match one_pass(&path, &owner, &counter, stale_after_ms)? {
569 Pass::Held(holder) => return Ok(DirLock::start(path, owner, counter, heartbeat_ms, &holder)),
570 Pass::Blocked(seen) => {
571 if let Some(holder) = seen {
572 last_seen = Some(holder.clone());
573 blocker = Some(holder);
574 }
575 }
576 Pass::GraceOwed(holder) => {
577 last_seen = Some(holder.clone());
578 let already_served = served.is_some_and(|served| served.names(&holder));
580 if already_served {
581 served = None;
582 }
583 let taken = if already_served {
584 finish_after_grace(&path, &holder, &owner, &counter)?
585 } else if deadline.saturating_duration_since(Instant::now()) >= grace {
588 thread::sleep(grace);
589 finish_after_grace(&path, &holder, &owner, &counter)?
590 } else {
591 None
592 };
593 if let Some(taken) = taken {
594 return Ok(DirLock::start(path, owner, counter, heartbeat_ms, &taken));
595 }
596 blocker = Some(holder);
597 }
598 Pass::Vacated => retry_now = true,
599 }
600
601 if retry_now && retries < MAX_IMMEDIATE_RETRIES {
602 retries += 1;
603 continue;
604 }
605
606 let remaining = deadline.saturating_duration_since(Instant::now());
607 if remaining.is_zero() {
608 return Err(DirLockError::Locked(LockedError {
609 subject,
610 holder: blocker.or(last_seen),
611 waited_ms: started_at.elapsed().as_millis() as u64,
612 }));
613 }
614 thread::sleep(remaining.min(Duration::from_millis(retry_interval_ms)));
615 }
616 }
617
618 pub fn challenge(path: impl Into<PathBuf>, options: LockOptions) -> Result<ChallengeOutcome, DirLockError> {
631 let path = path.into();
632 let stale_after_ms = options.stale_after_ms.unwrap_or(DEFAULT_STALE_MS);
633 let heartbeat_ms = options.heartbeat_ms.unwrap_or(DEFAULT_HEARTBEAT_MS);
634 let owner = mint_owner();
635 let counter = AtomicU64::new(0);
636
637 for _ in 0..=MAX_IMMEDIATE_RETRIES {
638 match one_pass(&path, &owner, &counter, stale_after_ms)? {
639 Pass::Held(holder) => {
640 let lock = DirLock::start(path, owner, counter, heartbeat_ms, &holder);
641 return Ok(ChallengeOutcome::Taken(lock));
642 }
643 Pass::Blocked(seen) => return Ok(ChallengeOutcome::Blocked(seen)),
644 Pass::GraceOwed(holder) => {
645 return Ok(ChallengeOutcome::GraceOwed {
646 holder,
647 grace_ms: stale_after_ms / 2,
648 })
649 }
650 Pass::Vacated => {}
651 }
652 }
653 Ok(ChallengeOutcome::Blocked(None))
654 }
655
656 pub fn finish_challenge(
670 path: impl Into<PathBuf>,
671 options: LockOptions,
672 holder: &LockHolder,
673 ) -> Result<Option<DirLock>, DirLockError> {
674 let path = path.into();
675 let heartbeat_ms = options.heartbeat_ms.unwrap_or(DEFAULT_HEARTBEAT_MS);
676 let owner = mint_owner();
677 let counter = AtomicU64::new(0);
678 let taken = finish_after_grace(&path, holder, &owner, &counter)?;
679 Ok(taken.map(|taken| DirLock::start(path, owner, counter, heartbeat_ms, &taken)))
680 }
681
682 pub fn inspect(path: impl AsRef<Path>, options: LockOptions) -> Result<LockState, DirLockError> {
695 let stale_after_ms = options.stale_after_ms.unwrap_or(DEFAULT_STALE_MS);
696 Ok(match read_lock_file(path.as_ref())? {
697 LockFileState::Missing => LockState::Free,
698 LockFileState::Unreadable { .. } => LockState::Unreadable,
699 LockFileState::Holder(holder) => match abandonment(&holder, stale_after_ms) {
700 Abandonment::Held => LockState::Held(holder),
701 Abandonment::LeaseExpired => LockState::LeaseExpired(holder),
702 Abandonment::ProvablyDead => LockState::HolderGone(holder),
703 },
704 })
705 }
706
707 fn start(path: PathBuf, owner: String, counter: AtomicU64, heartbeat_ms: u64, holder: &LockHolder) -> DirLock {
708 let beat = Arc::new(Beat {
709 path,
710 owner,
711 heartbeat_ms,
712 counter,
713 held: AtomicBool::new(true),
714 lost: AtomicBool::new(false),
715 stop: AtomicBool::new(false),
716 last_beat_ms: AtomicI64::new(holder.heartbeat_at),
717 });
718
719 #[cfg(unix)]
720 let heartbeat = spawn_heartbeat(Arc::clone(&beat), heartbeat_ms);
721
722 DirLock {
723 beat,
724 #[cfg(unix)]
725 heartbeat,
726 }
727 }
728
729 pub fn tick(&self) -> Result<(), DirLockError> {
737 self.beat.renew_if_due()
738 }
739
740 pub fn is_lost(&self) -> bool {
743 self.beat.lost.load(Ordering::SeqCst)
744 }
745
746 pub fn release(&mut self) {
747 if !self.beat.held.swap(false, Ordering::SeqCst) {
748 return;
749 }
750 self.beat.stop.store(true, Ordering::SeqCst);
751 #[cfg(unix)]
752 if let Some(handle) = self.heartbeat.take() {
753 let _ = handle.join();
754 }
755 if self.beat.lost.load(Ordering::SeqCst) {
756 return;
757 }
758 if let Ok(LockFileState::Holder(holder)) = read_lock_file(&self.beat.path) {
761 if holder.owner == self.beat.owner {
762 let _ = fs::remove_file(&self.beat.path);
763 }
764 }
765 }
766
767 #[cfg(test)]
768 fn owner(&self) -> &str {
769 &self.beat.owner
770 }
771
772 #[cfg(test)]
773 fn temp_path_at(&self, ordinal: u64) -> PathBuf {
774 temp_path_for(&self.beat.path, &self.beat.owner, ordinal)
775 }
776}
777
778#[derive(Debug)]
780pub enum ChallengeOutcome {
781 Taken(DirLock),
784 Blocked(Option<LockHolder>),
787 GraceOwed { holder: LockHolder, grace_ms: u64 },
791}
792
793#[derive(Debug, Clone, PartialEq, Eq)]
795pub enum LockState {
796 Free,
798 Held(LockHolder),
800 LeaseExpired(LockHolder),
803 HolderGone(LockHolder),
807 Unreadable,
811}
812
813impl LockState {
814 pub fn holder(&self) -> Option<&LockHolder> {
817 match self {
818 LockState::Held(holder) | LockState::LeaseExpired(holder) | LockState::HolderGone(holder) => Some(holder),
819 LockState::Free | LockState::Unreadable => None,
820 }
821 }
822
823 pub fn blocks_a_writer(&self) -> bool {
828 match self {
829 LockState::Held(_) | LockState::LeaseExpired(_) | LockState::Unreadable => true,
830 LockState::Free | LockState::HolderGone(_) => false,
831 }
832 }
833}
834
835#[cfg(unix)]
838fn spawn_heartbeat(beat: Arc<Beat>, heartbeat_ms: u64) -> Option<thread::JoinHandle<()>> {
839 let step = Duration::from_millis(heartbeat_ms.clamp(1, HEARTBEAT_STEP_MS));
840 let ticks_per_beat = (heartbeat_ms / step.as_millis().max(1) as u64).max(1);
841 thread::Builder::new()
842 .name("hayson-lock-heartbeat".to_string())
843 .spawn(move || loop {
844 for _ in 0..ticks_per_beat {
845 if beat.stop.load(Ordering::SeqCst) {
846 return;
847 }
848 thread::sleep(step);
849 }
850 if beat.stop.load(Ordering::SeqCst) {
851 return;
852 }
853 if matches!(beat.renew(), Err(DirLockError::Lost(_))) {
855 return;
856 }
857 })
858 .ok()
859}
860
861impl Drop for DirLock {
862 fn drop(&mut self) {
863 self.beat.stop.store(true, Ordering::SeqCst);
865 }
866}
867
868#[cfg(test)]
869mod tests {
870 use super::*;
871 use crate::lock::platform::tests::{with_pid_liveness, without_pid_liveness};
872 use std::os::unix::fs::symlink;
873
874 fn temp_dir(prefix: &str) -> PathBuf {
875 let dir = std::env::temp_dir().join(format!(
876 "{prefix}{}-{}",
877 std::process::id(),
878 SystemTime::now().duration_since(UNIX_EPOCH).unwrap().as_nanos()
879 ));
880 fs::create_dir_all(&dir).unwrap();
881 dir
882 }
883
884 fn holder_of(path: &Path) -> LockHolder {
885 serde_json::from_str(&fs::read_to_string(path).unwrap()).unwrap()
886 }
887
888 fn until<T>(mut seen: impl FnMut() -> Option<T>, within_ms: u64) -> Option<T> {
889 let deadline = Instant::now() + Duration::from_millis(within_ms);
890 loop {
891 if let Some(found) = seen() {
892 return Some(found);
893 }
894 if Instant::now() >= deadline {
895 return None;
896 }
897 thread::sleep(Duration::from_millis(10));
898 }
899 }
900
901 #[test]
902 fn writes_the_holder_and_removes_the_file_on_release() {
903 let dir = temp_dir("hayson-lock-basic-");
904 let path = dir.join(LOCK_FILE);
905 let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();
906
907 let holder = holder_of(&path);
908 assert_eq!(holder.owner, lock.owner());
909 assert_eq!(holder.pid, Some(std::process::id()));
910 assert!(holder.host.is_some());
911
912 lock.release();
913 assert!(!path.exists());
914 }
915
916 #[test]
917 fn refuses_a_second_writer_while_the_first_is_alive() {
918 let dir = temp_dir("hayson-lock-contend-");
919 let path = dir.join(LOCK_FILE);
920 let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();
921
922 let second = DirLock::acquire(
923 path.clone(),
924 LockOptions {
925 timeout_ms: Some(0),
926 ..Default::default()
927 },
928 );
929 assert!(matches!(second, Err(DirLockError::Locked(_))));
930
931 lock.release();
932 DirLock::acquire(path, LockOptions::default()).unwrap().release();
933 }
934
935 #[test]
936 fn a_clean_release_frees_the_directory_at_once_without_waiting_out_the_lease() {
937 let dir = temp_dir("hayson-lock-clean-release-");
938 let path = dir.join(LOCK_FILE);
939 let options = LockOptions {
940 stale_after_ms: Some(60_000),
941 timeout_ms: Some(0),
942 ..Default::default()
943 };
944 let mut first = DirLock::acquire(path.clone(), options).unwrap();
945 first.release();
946
947 let started_at = Instant::now();
948 let mut second = DirLock::acquire(path, options).unwrap();
949 assert!(started_at.elapsed() < Duration::from_millis(500));
950 second.release();
951 }
952
953 #[test]
954 fn waits_up_to_the_timeout_and_takes_the_lock_once_released() {
955 let dir = temp_dir("hayson-lock-wait-");
956 let path = dir.join(LOCK_FILE);
957 let mut first = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();
958
959 thread::spawn(move || {
960 thread::sleep(Duration::from_millis(150));
961 first.release();
962 });
963
964 let started_at = Instant::now();
965 let mut second = DirLock::acquire(
966 path,
967 LockOptions {
968 timeout_ms: Some(5_000),
969 retry_interval_ms: Some(10),
970 ..Default::default()
971 },
972 )
973 .unwrap();
974 assert!(started_at.elapsed() >= Duration::from_millis(100));
975 second.release();
976 }
977
978 #[test]
979 fn reports_how_long_it_waited_before_giving_up() {
980 let dir = temp_dir("hayson-lock-report-");
981 let path = dir.join(LOCK_FILE);
982 let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();
983
984 let failure = DirLock::acquire(
985 path,
986 LockOptions {
987 timeout_ms: Some(200),
988 retry_interval_ms: Some(10),
989 ..Default::default()
990 },
991 )
992 .unwrap_err();
993 match failure {
994 DirLockError::Locked(error) => {
995 assert!(error.waited_ms >= 180);
996 assert_eq!(error.holder.unwrap().owner, lock.owner());
997 }
998 other => panic!("expected Locked, got {other}"),
999 }
1000 lock.release();
1001 }
1002
1003 #[test]
1004 fn reclaims_a_lock_whose_owner_is_gone() {
1005 let _pids = with_pid_liveness();
1008 let dir = temp_dir("hayson-lock-dead-owner-");
1009 let path = dir.join(LOCK_FILE);
1010 let dead = LockHolder {
1011 owner: mint_owner(),
1012 pid: Some(2u32.pow(22)),
1013 host: platform::hostname(),
1014 started_at: now_ms(),
1015 heartbeat_at: now_ms(),
1016 };
1017 fs::write(&path, serde_json::to_string(&dead).unwrap()).unwrap();
1018
1019 let mut lock = DirLock::acquire(
1020 path.clone(),
1021 LockOptions {
1022 stale_after_ms: Some(60_000),
1023 timeout_ms: Some(0),
1024 ..Default::default()
1025 },
1026 )
1027 .unwrap();
1028 assert_eq!(holder_of(&path).owner, lock.owner());
1029 lock.release();
1030 }
1031
1032 #[test]
1035 fn reclaims_an_expired_lease_on_the_lease_alone_with_no_pid_liveness_to_help() {
1036 let _no_pids = without_pid_liveness();
1037 let dir = temp_dir("hayson-lock-lease-only-");
1038 let path = dir.join(LOCK_FILE);
1039 let stale = LockHolder {
1040 owner: mint_owner(),
1041 pid: Some(std::process::id()),
1043 host: platform::hostname(),
1044 started_at: 0,
1045 heartbeat_at: 0,
1046 };
1047 fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();
1048
1049 let mut lock = DirLock::acquire(
1050 path.clone(),
1051 LockOptions {
1052 stale_after_ms: Some(120),
1053 timeout_ms: Some(2_000),
1054 retry_interval_ms: Some(10),
1055 ..Default::default()
1056 },
1057 )
1058 .unwrap();
1059 assert_eq!(holder_of(&path).owner, lock.owner());
1060 lock.release();
1061 }
1062
1063 #[test]
1064 fn reclaims_a_lock_whose_heartbeat_went_stale() {
1065 let dir = temp_dir("hayson-lock-stale-heartbeat-");
1066 let path = dir.join(LOCK_FILE);
1067 let stale = LockHolder {
1068 owner: mint_owner(),
1069 pid: Some(std::process::id()),
1070 host: Some("some-other-host".to_string()),
1071 started_at: 0,
1072 heartbeat_at: 0,
1073 };
1074 fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();
1075
1076 let mut lock = DirLock::acquire(
1077 path,
1078 LockOptions {
1079 stale_after_ms: Some(1),
1080 ..Default::default()
1081 },
1082 )
1083 .unwrap();
1084 lock.release();
1085 }
1086
1087 #[test]
1088 fn inspect_names_the_live_holder_and_leaves_the_lock_where_it_is() {
1089 let dir = temp_dir("hayson-lock-inspect-held-");
1090 let path = dir.join(LOCK_FILE);
1091 let mut lock = DirLock::acquire(path.clone(), LockOptions::default()).unwrap();
1092
1093 let state = DirLock::inspect(&path, LockOptions::default()).unwrap();
1094
1095 match &state {
1096 LockState::Held(holder) => {
1097 assert_eq!(holder.owner, lock.owner());
1098 assert_eq!(holder.pid, Some(std::process::id()));
1099 }
1100 other => panic!("expected the live holder, got {other:?}"),
1101 }
1102 assert!(state.blocks_a_writer());
1103 assert_eq!(state.holder().map(|holder| holder.owner.as_str()), Some(lock.owner()));
1104 assert_eq!(holder_of(&path).owner, lock.owner());
1106 lock.release();
1107 }
1108
1109 #[test]
1110 fn inspect_reads_a_directory_nobody_holds_as_free_without_creating_a_lock_file() {
1111 let dir = temp_dir("hayson-lock-inspect-free-");
1112 let path = dir.join(LOCK_FILE);
1113
1114 let state = DirLock::inspect(&path, LockOptions::default()).unwrap();
1115
1116 assert_eq!(state, LockState::Free);
1117 assert!(!state.blocks_a_writer());
1118 assert!(!path.exists(), "inspecting must not plant a lock file");
1119 }
1120
1121 #[test]
1122 fn inspect_tells_an_expired_lease_from_a_holder_that_is_provably_gone() {
1123 let _pids = with_pid_liveness();
1124 let dir = temp_dir("hayson-lock-inspect-abandoned-");
1125 let stale_path = dir.join(LOCK_FILE);
1126 let dead_path = dir.join("dead.lock");
1127 let holder = |pid: u32| LockHolder {
1128 owner: mint_owner(),
1129 pid: Some(pid),
1130 host: platform::hostname(),
1131 started_at: 0,
1132 heartbeat_at: 0,
1133 };
1134 fs::write(&stale_path, serde_json::to_string(&holder(std::process::id())).unwrap()).unwrap();
1135 fs::write(&dead_path, serde_json::to_string(&holder(2u32.pow(22))).unwrap()).unwrap();
1136 let options = LockOptions {
1137 stale_after_ms: Some(1),
1138 ..Default::default()
1139 };
1140
1141 let stale = DirLock::inspect(&stale_path, options).unwrap();
1142 let dead = DirLock::inspect(&dead_path, options).unwrap();
1143
1144 assert!(matches!(stale, LockState::LeaseExpired(_)));
1145 assert!(matches!(dead, LockState::HolderGone(_)));
1146 assert!(stale.blocks_a_writer(), "the hand-over grace is still owed");
1147 assert!(!dead.blocks_a_writer(), "nothing is left there to give notice to");
1148 }
1149
1150 #[test]
1151 fn inspect_reads_a_lock_file_it_cannot_parse_as_unreadable_and_refuses_a_planted_link() {
1152 let dir = temp_dir("hayson-lock-inspect-unreadable-");
1153 let path = dir.join(LOCK_FILE);
1154 let linked = dir.join("linked.lock");
1155 fs::write(&path, "{ not a holder }").unwrap();
1156 symlink(dir.join("elsewhere"), &linked).unwrap();
1157
1158 let unreadable = DirLock::inspect(&path, LockOptions::default()).unwrap();
1159 let planted = DirLock::inspect(&linked, LockOptions::default());
1160
1161 assert_eq!(unreadable, LockState::Unreadable);
1162 assert!(unreadable.blocks_a_writer(), "it is reclaimed only once it goes stale");
1163 assert!(matches!(planted, Err(DirLockError::SuspiciousLockFile(_))));
1164 }
1165
1166 #[test]
1167 fn never_steals_from_a_holder_that_keeps_beating_across_several_lease_periods() {
1168 let _no_pids = without_pid_liveness();
1169 let dir = temp_dir("hayson-lock-no-false-steal-");
1170 let path = dir.join(LOCK_FILE);
1171 let mut held = DirLock::acquire(
1172 path.clone(),
1173 LockOptions {
1174 stale_after_ms: Some(120),
1175 heartbeat_ms: Some(20),
1176 ..Default::default()
1177 },
1178 )
1179 .unwrap();
1180 let owner = held.owner().to_string();
1181
1182 for _ in 0..4 {
1183 let refused = DirLock::acquire(
1184 path.clone(),
1185 LockOptions {
1186 stale_after_ms: Some(120),
1187 timeout_ms: Some(150),
1188 retry_interval_ms: Some(10),
1189 ..Default::default()
1190 },
1191 );
1192 assert!(matches!(refused, Err(DirLockError::Locked(_))));
1193 assert_eq!(holder_of(&path).owner, owner);
1194 }
1195
1196 assert!(held.tick().is_ok());
1197 assert!(!held.is_lost());
1198 held.release();
1199 }
1200
1201 #[test]
1204 fn a_holder_that_resumes_beating_inside_the_grace_window_keeps_its_lock() {
1205 let _no_pids = without_pid_liveness();
1206 let dir = temp_dir("hayson-lock-grace-recovery-");
1207 let path = dir.join(LOCK_FILE);
1208 let owner = mint_owner();
1209 let counter = AtomicU64::new(0);
1210 let stale = LockHolder {
1211 owner: owner.clone(),
1212 pid: Some(std::process::id()),
1213 host: platform::hostname(),
1214 started_at: 0,
1215 heartbeat_at: 0,
1216 };
1217 fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();
1218
1219 let beating_path = path.clone();
1220 let beating_owner = owner.clone();
1221 let resumed = thread::spawn(move || {
1222 thread::sleep(Duration::from_millis(150));
1223 let beaten = LockHolder {
1224 heartbeat_at: now_ms(),
1225 ..stale
1226 };
1227 let temp = write_temp(&beating_path, &beating_owner, &counter, &beaten).unwrap();
1228 fs::rename(&temp, &beating_path).unwrap();
1229 });
1230
1231 let refused = DirLock::acquire(
1232 path.clone(),
1233 LockOptions {
1234 stale_after_ms: Some(600),
1235 timeout_ms: Some(400),
1236 retry_interval_ms: Some(10),
1237 ..Default::default()
1238 },
1239 );
1240 resumed.join().unwrap();
1241 assert!(matches!(refused, Err(DirLockError::Locked(_))));
1242 assert_eq!(holder_of(&path).owner, owner);
1243 }
1244
1245 #[test]
1246 fn a_dispossessed_holder_learns_it_lost_the_lock_on_its_next_tick() {
1247 let _no_pids = without_pid_liveness();
1248 let dir = temp_dir("hayson-lock-fencing-");
1249 let path = dir.join(LOCK_FILE);
1250 let mut robbed = DirLock::acquire(
1251 path.clone(),
1252 LockOptions {
1253 stale_after_ms: Some(120),
1254 heartbeat_ms: Some(60_000),
1257 ..Default::default()
1258 },
1259 )
1260 .unwrap();
1261 assert!(robbed.tick().is_ok());
1262
1263 thread::sleep(Duration::from_millis(200));
1264 let mut thief = DirLock::acquire(
1265 path.clone(),
1266 LockOptions {
1267 stale_after_ms: Some(120),
1268 timeout_ms: Some(2_000),
1269 retry_interval_ms: Some(10),
1270 ..Default::default()
1271 },
1272 )
1273 .unwrap();
1274 assert_ne!(thief.owner(), robbed.owner());
1275
1276 robbed.beat.last_beat_ms.store(0, Ordering::SeqCst);
1279 let fenced = robbed.tick();
1280 assert!(
1281 matches!(fenced, Err(DirLockError::Lost(_))),
1282 "expected Lost, got {fenced:?}"
1283 );
1284 assert!(robbed.is_lost());
1285 assert!(matches!(robbed.tick(), Err(DirLockError::Lost(_))));
1286
1287 robbed.release();
1289 assert_eq!(holder_of(&path).owner, thief.owner());
1290 thief.release();
1291 }
1292
1293 #[test]
1294 fn exactly_one_of_many_concurrent_challengers_takes_an_expired_lock() {
1295 let _no_pids = without_pid_liveness();
1296 let dir = temp_dir("hayson-lock-steal-race-");
1297 let path = dir.join(LOCK_FILE);
1298 let stale = LockHolder {
1299 owner: mint_owner(),
1300 pid: Some(std::process::id()),
1301 host: platform::hostname(),
1302 started_at: 0,
1303 heartbeat_at: 0,
1304 };
1305 fs::write(&path, serde_json::to_string(&stale).unwrap()).unwrap();
1306
1307 let mut challengers = Vec::new();
1308 for _ in 0..8 {
1309 let path = path.clone();
1310 challengers.push(thread::spawn(move || {
1311 DirLock::acquire(
1312 path,
1313 LockOptions {
1314 stale_after_ms: Some(200),
1315 heartbeat_ms: Some(60_000),
1316 timeout_ms: Some(400),
1317 retry_interval_ms: Some(5),
1318 grace_served: None,
1319 },
1320 )
1321 }));
1322 }
1323
1324 let mut winners = Vec::new();
1325 for challenger in challengers {
1326 match challenger.join().unwrap() {
1327 Ok(lock) => winners.push(lock),
1328 Err(DirLockError::Locked(_)) => {}
1329 Err(other) => panic!("unexpected failure: {other}"),
1330 }
1331 }
1332
1333 assert!(!winners.is_empty(), "an expired lock was left to nobody");
1338 let landed = holder_of(&path).owner;
1339 let mut holding = 0;
1340 for winner in &mut winners {
1341 winner.beat.last_beat_ms.store(0, Ordering::SeqCst);
1342 if winner.owner() == landed {
1343 holding += 1;
1344 assert!(winner.tick().is_ok(), "the writer named in the lock file was fenced");
1345 } else {
1346 assert!(
1347 matches!(winner.tick(), Err(DirLockError::Lost(_))),
1348 "a challenger that did not land in the lock file was not fenced",
1349 );
1350 }
1351 }
1352 assert_eq!(holding, 1, "more than one challenger owns the same lock file");
1353
1354 for winner in &mut winners {
1355 winner.release();
1356 }
1357 assert!(!path.exists(), "the surviving writer did not clean up after itself");
1358 }
1359
1360 fn abandoned_lease(path: &Path) -> LockHolder {
1362 let stale = LockHolder {
1363 owner: mint_owner(),
1364 pid: Some(std::process::id()),
1366 host: platform::hostname(),
1367 started_at: 0,
1368 heartbeat_at: 0,
1369 };
1370 fs::write(path, serde_json::to_string(&stale).unwrap()).unwrap();
1371 stale
1372 }
1373
1374 fn grace_owed(outcome: ChallengeOutcome) -> (LockHolder, u64) {
1375 match outcome {
1376 ChallengeOutcome::GraceOwed { holder, grace_ms } => (holder, grace_ms),
1377 other => panic!("expected GraceOwed, got {other:?}"),
1378 }
1379 }
1380
1381 #[test]
1384 fn challenge_reports_the_grace_an_expired_lease_is_owed_without_waiting_any_of_it() {
1385 let _no_pids = without_pid_liveness();
1386 let dir = temp_dir("hayson-lock-challenge-");
1387 let path = dir.join(LOCK_FILE);
1388 let stale = abandoned_lease(&path);
1389 let options = LockOptions {
1390 stale_after_ms: Some(4_000),
1391 ..Default::default()
1392 };
1393
1394 let started_at = Instant::now();
1395 let (holder, grace_ms) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
1396 let answered_in = started_at.elapsed();
1397
1398 assert_eq!(holder.owner, stale.owner);
1399 assert_eq!(grace_ms, 2_000);
1400 assert!(
1401 answered_in < Duration::from_millis(500),
1402 "the challenge waited {answered_in:?} of the grace it was supposed to hand back",
1403 );
1404
1405 thread::sleep(Duration::from_millis(grace_ms.min(120)));
1407 let mut lock = DirLock::finish_challenge(path.clone(), options, &holder)
1408 .unwrap()
1409 .unwrap();
1410 assert_eq!(holder_of(&path).owner, lock.owner());
1411 lock.release();
1412 }
1413
1414 #[test]
1417 fn challenge_takes_a_free_lock_outright_and_reports_a_live_holder_as_blocked() {
1418 let dir = temp_dir("hayson-lock-challenge-free-");
1419 let path = dir.join(LOCK_FILE);
1420 let options = LockOptions::default();
1421
1422 let taken = DirLock::challenge(path.clone(), options).unwrap();
1423 let mut first = match taken {
1424 ChallengeOutcome::Taken(lock) => lock,
1425 other => panic!("expected Taken, got {other:?}"),
1426 };
1427
1428 match DirLock::challenge(path.clone(), options).unwrap() {
1429 ChallengeOutcome::Blocked(Some(holder)) => assert_eq!(holder.owner, first.owner()),
1430 other => panic!("expected Blocked, got {other:?}"),
1431 }
1432 first.release();
1433 }
1434
1435 #[test]
1438 fn of_two_callers_finishing_one_challenge_exactly_one_takes_the_lock_and_the_other_is_refused() {
1439 let _no_pids = without_pid_liveness();
1440 let dir = temp_dir("hayson-lock-finish-race-");
1441 let path = dir.join(LOCK_FILE);
1442 abandoned_lease(&path);
1443 let options = LockOptions {
1444 stale_after_ms: Some(200),
1445 heartbeat_ms: Some(60_000),
1446 ..Default::default()
1447 };
1448
1449 let (first_seen, _) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
1450 let (second_seen, _) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
1451 assert_eq!(first_seen, second_seen, "both challengers read the same lease");
1452
1453 thread::sleep(Duration::from_millis(120));
1454 let winner = DirLock::finish_challenge(path.clone(), options, &first_seen).unwrap();
1455 let loser = DirLock::finish_challenge(path.clone(), options, &second_seen).unwrap();
1456
1457 let mut winner = winner.expect("the first caller to finish the challenge got nothing");
1458 assert!(
1459 loser.is_none(),
1460 "the second caller took a lock the first had already reclaimed",
1461 );
1462 assert_eq!(holder_of(&path).owner, winner.owner());
1463 winner.release();
1464 assert!(!path.exists(), "the surviving writer did not clean up after itself");
1465 }
1466
1467 #[test]
1471 fn concurrent_finishers_of_one_challenge_leave_the_lock_file_naming_exactly_one_of_them() {
1472 let _no_pids = without_pid_liveness();
1473 let dir = temp_dir("hayson-lock-finish-threads-");
1474 let path = dir.join(LOCK_FILE);
1475 abandoned_lease(&path);
1476 let options = LockOptions {
1477 stale_after_ms: Some(200),
1478 heartbeat_ms: Some(60_000),
1479 ..Default::default()
1480 };
1481
1482 let mut finishers = Vec::new();
1483 for _ in 0..8 {
1484 let path = path.clone();
1485 finishers.push(thread::spawn(move || {
1486 let (holder, _) = match DirLock::challenge(path.clone(), options).unwrap() {
1487 ChallengeOutcome::GraceOwed { holder, grace_ms } => (holder, grace_ms),
1488 other => panic!("expected GraceOwed, got {other:?}"),
1489 };
1490 thread::sleep(Duration::from_millis(120));
1491 DirLock::finish_challenge(path, options, &holder).unwrap()
1492 }));
1493 }
1494
1495 let mut winners: Vec<DirLock> = finishers
1496 .into_iter()
1497 .filter_map(|finisher| finisher.join().unwrap())
1498 .collect();
1499
1500 assert!(!winners.is_empty(), "an expired lock was left to nobody");
1501 let landed = holder_of(&path).owner;
1502 let mut holding = 0;
1503 for winner in &mut winners {
1504 winner.beat.last_beat_ms.store(0, Ordering::SeqCst);
1505 if winner.owner() == landed {
1506 holding += 1;
1507 assert!(winner.tick().is_ok(), "the writer named in the lock file was fenced");
1508 } else {
1509 assert!(
1510 matches!(winner.tick(), Err(DirLockError::Lost(_))),
1511 "a finisher that did not land in the lock file was not fenced",
1512 );
1513 }
1514 }
1515 assert_eq!(holding, 1, "more than one finisher owns the same lock file");
1516
1517 for winner in &mut winners {
1518 winner.release();
1519 }
1520 assert!(!path.exists(), "the surviving writer did not clean up after itself");
1521 }
1522
1523 #[test]
1526 fn finishing_a_challenge_a_resumed_holder_has_moved_past_reclaims_nothing() {
1527 let _no_pids = without_pid_liveness();
1528 let dir = temp_dir("hayson-lock-finish-resumed-");
1529 let path = dir.join(LOCK_FILE);
1530 let stale = abandoned_lease(&path);
1531 let options = LockOptions {
1532 stale_after_ms: Some(600),
1533 ..Default::default()
1534 };
1535
1536 let (holder, _) = grace_owed(DirLock::challenge(path.clone(), options).unwrap());
1537
1538 let counter = AtomicU64::new(0);
1539 let beaten = LockHolder {
1540 heartbeat_at: now_ms(),
1541 ..stale.clone()
1542 };
1543 let temp = write_temp(&path, &stale.owner, &counter, &beaten).unwrap();
1544 fs::rename(&temp, &path).unwrap();
1545
1546 assert!(
1547 DirLock::finish_challenge(path.clone(), options, &holder)
1548 .unwrap()
1549 .is_none(),
1550 "a holder that proved itself alive inside the grace lost its lock anyway",
1551 );
1552 assert_eq!(holder_of(&path).owner, stale.owner);
1553 }
1554
1555 #[test]
1559 fn a_fail_fast_acquire_carrying_a_served_grace_reclaims_the_lease_that_token_names() {
1560 let _no_pids = without_pid_liveness();
1561 let dir = temp_dir("hayson-lock-served-");
1562 let path = dir.join(LOCK_FILE);
1563 abandoned_lease(&path);
1564 let looking = LockOptions {
1565 stale_after_ms: Some(4_000),
1566 timeout_ms: Some(0),
1567 ..Default::default()
1568 };
1569
1570 let refused = DirLock::acquire(path.clone(), looking).unwrap_err();
1572 assert!(matches!(refused, DirLockError::Locked(_)), "got {refused:?}");
1573
1574 let (holder, grace_ms) = grace_owed(DirLock::challenge(path.clone(), looking).unwrap());
1575 assert_eq!(grace_ms, 2_000);
1576 thread::sleep(Duration::from_millis(120));
1577
1578 let started_at = Instant::now();
1579 let mut lock = DirLock::acquire(
1580 path.clone(),
1581 LockOptions {
1582 grace_served: GraceServed::of(&holder),
1583 ..looking
1584 },
1585 )
1586 .unwrap();
1587 assert!(
1588 started_at.elapsed() < Duration::from_millis(500),
1589 "the open slept a grace its caller had already served",
1590 );
1591 assert_eq!(holder_of(&path).owner, lock.owner());
1592 lock.release();
1593 }
1594
1595 #[test]
1598 fn a_served_grace_naming_another_lease_reclaims_nothing() {
1599 let _no_pids = without_pid_liveness();
1600 let dir = temp_dir("hayson-lock-served-stale-token-");
1601 let path = dir.join(LOCK_FILE);
1602 let first = abandoned_lease(&path);
1603 let served = GraceServed::of(&first);
1604 assert!(served.is_some());
1605
1606 let second = abandoned_lease(&path);
1608 assert_ne!(second.owner, first.owner);
1609
1610 let refused = DirLock::acquire(
1611 path.clone(),
1612 LockOptions {
1613 stale_after_ms: Some(4_000),
1614 timeout_ms: Some(0),
1615 grace_served: served,
1616 ..Default::default()
1617 },
1618 )
1619 .unwrap_err();
1620 assert!(matches!(refused, DirLockError::Locked(_)), "got {refused:?}");
1621 assert_eq!(holder_of(&path).owner, second.owner);
1622 }
1623
1624 #[test]
1625 fn a_served_grace_is_only_read_off_an_owner_token_this_engine_minted() {
1626 let minted = mint_owner();
1627 assert!(GraceServed::parse(&minted, 7).is_some());
1628 assert_eq!(GraceServed::parse(&minted, 7), GraceServed::parse(&minted, 7));
1629 assert_ne!(GraceServed::parse(&minted, 7), GraceServed::parse(&minted, 8));
1630 assert!(GraceServed::parse("", 7).is_none());
1631 assert!(GraceServed::parse("not-a-token", 7).is_none());
1632 assert!(GraceServed::parse(&minted[..31], 7).is_none());
1633 assert!(GraceServed::parse("zz000000000000000000000000000000", 7).is_none());
1634 }
1635
1636 #[test]
1637 fn reclaims_an_unparsable_lock_file_nobody_touched_for_longer_than_stale_after_ms() {
1638 let dir = temp_dir("hayson-lock-unparsable-stale-");
1639 let path = dir.join(LOCK_FILE);
1640 fs::write(&path, "{\"pid\":48213,\"host\":\"thi").unwrap();
1641 let past = SystemTime::now() - Duration::from_secs(60);
1642 let file = fs::File::options().write(true).open(&path).unwrap();
1643 file.set_times(fs::FileTimes::new().set_modified(past)).unwrap();
1644
1645 let mut lock = DirLock::acquire(
1646 path.clone(),
1647 LockOptions {
1648 stale_after_ms: Some(1_000),
1649 timeout_ms: Some(200),
1650 ..Default::default()
1651 },
1652 )
1653 .unwrap();
1654 assert_eq!(holder_of(&path).owner, lock.owner());
1655 lock.release();
1656 }
1657
1658 #[test]
1661 fn an_old_format_pid_only_lock_file_is_unparsable_and_reclaimed_after_the_lease() {
1662 let dir = temp_dir("hayson-lock-old-format-");
1663 let path = dir.join(LOCK_FILE);
1664 let old = serde_json::json!({
1665 "pid": std::process::id(),
1666 "host": platform::hostname(),
1667 "started_at": now_ms(),
1668 "heartbeat_at": now_ms(),
1669 });
1670 fs::write(&path, serde_json::to_string(&old).unwrap()).unwrap();
1671 assert!(serde_json::from_str::<LockHolder>(&fs::read_to_string(&path).unwrap()).is_err());
1672
1673 let fresh = DirLock::acquire(
1674 path.clone(),
1675 LockOptions {
1676 stale_after_ms: Some(30_000),
1677 timeout_ms: Some(100),
1678 retry_interval_ms: Some(10),
1679 ..Default::default()
1680 },
1681 );
1682 assert!(
1683 matches!(fresh, Err(DirLockError::Locked(_))),
1684 "a fresh unparsable file is left alone"
1685 );
1686
1687 let past = SystemTime::now() - Duration::from_secs(60);
1688 let file = fs::File::options().write(true).open(&path).unwrap();
1689 file.set_times(fs::FileTimes::new().set_modified(past)).unwrap();
1690
1691 let mut lock = DirLock::acquire(
1692 path.clone(),
1693 LockOptions {
1694 stale_after_ms: Some(1_000),
1695 timeout_ms: Some(200),
1696 ..Default::default()
1697 },
1698 )
1699 .unwrap();
1700 assert_eq!(holder_of(&path).owner, lock.owner());
1701 lock.release();
1702 }
1703
1704 #[test]
1705 fn leaves_a_freshly_written_unparsable_lock_file_alone() {
1706 let dir = temp_dir("hayson-lock-unparsable-fresh-");
1707 let path = dir.join(LOCK_FILE);
1708 fs::write(&path, "").unwrap();
1709
1710 let failure = DirLock::acquire(
1711 path,
1712 LockOptions {
1713 stale_after_ms: Some(30_000),
1714 timeout_ms: Some(100),
1715 retry_interval_ms: Some(10),
1716 ..Default::default()
1717 },
1718 )
1719 .unwrap_err();
1720 assert!(matches!(failure, DirLockError::Locked(_)));
1721 }
1722
1723 #[test]
1724 fn refuses_to_touch_a_lock_path_somebody_replaced_with_a_symlink() {
1725 let dir = temp_dir("hayson-lock-symlink-");
1726 let victim_dir = temp_dir("hayson-lock-victim-");
1727 let victim = victim_dir.join("precious.txt");
1728 fs::write(&victim, "PRECIOUS\n").unwrap();
1729 let victim_mtime = fs::metadata(&victim).unwrap().modified().unwrap();
1730
1731 let path = dir.join(LOCK_FILE);
1732 symlink(&victim, &path).unwrap();
1733
1734 let failure = DirLock::acquire(
1735 path.clone(),
1736 LockOptions {
1737 stale_after_ms: Some(0),
1738 timeout_ms: Some(100),
1739 ..Default::default()
1740 },
1741 )
1742 .unwrap_err();
1743 assert!(
1744 matches!(failure, DirLockError::SuspiciousLockFile(ref at) if at == &path),
1745 "expected SuspiciousLockFile, got {failure:?}",
1746 );
1747
1748 assert_eq!(fs::read_to_string(&victim).unwrap(), "PRECIOUS\n");
1749 assert_eq!(fs::metadata(&victim).unwrap().modified().unwrap(), victim_mtime);
1750 assert!(fs::symlink_metadata(&path).unwrap().file_type().is_symlink());
1751 }
1752
1753 #[test]
1757 fn the_heartbeat_refuses_every_planted_temporary_and_beats_on_a_free_name() {
1758 let dir = temp_dir("hayson-heartbeat-symlink-");
1759 let path = dir.join(LOCK_FILE);
1760 let mut lock = DirLock::acquire(
1761 path.clone(),
1762 LockOptions {
1763 heartbeat_ms: Some(60_000),
1764 ..Default::default()
1765 },
1766 )
1767 .unwrap();
1768
1769 let victim_dir = temp_dir("hayson-heartbeat-victim-");
1770 let victim = victim_dir.join("precious.txt");
1771 fs::write(&victim, "PRECIOUS\n").unwrap();
1772 let victim_mtime = fs::metadata(&victim).unwrap().modified().unwrap();
1773
1774 let planted: Vec<PathBuf> = (0..3).map(|n| lock.temp_path_at(n)).collect();
1775 for temp in &planted {
1776 symlink(&victim, temp).unwrap();
1777 }
1778
1779 let before = holder_of(&path).heartbeat_at;
1780 thread::sleep(Duration::from_millis(2));
1781 lock.beat.renew().unwrap();
1782
1783 assert_eq!(fs::read_to_string(&victim).unwrap(), "PRECIOUS\n");
1784 assert_eq!(fs::metadata(&victim).unwrap().modified().unwrap(), victim_mtime);
1785 for temp in &planted {
1786 assert!(
1787 fs::symlink_metadata(temp).unwrap().file_type().is_symlink(),
1788 "a planted link was consumed instead of skipped",
1789 );
1790 }
1791 assert!(holder_of(&path).heartbeat_at > before);
1792 assert_eq!(holder_of(&path).owner, lock.owner());
1793 assert!(!fs::symlink_metadata(&path).unwrap().file_type().is_symlink());
1794 assert!(
1795 !lock.temp_path_at(3).exists(),
1796 "the temporary must not outlive the rename"
1797 );
1798
1799 lock.release();
1800 for temp in &planted {
1801 let _ = fs::remove_file(temp);
1802 }
1803 }
1804
1805 #[test]
1806 fn the_background_heartbeat_keeps_the_lease_fresh_on_its_own() {
1807 let dir = temp_dir("hayson-heartbeat-runs-");
1808 let path = dir.join(LOCK_FILE);
1809 let mut lock = DirLock::acquire(
1810 path.clone(),
1811 LockOptions {
1812 heartbeat_ms: Some(20),
1813 ..Default::default()
1814 },
1815 )
1816 .unwrap();
1817
1818 let before = holder_of(&path).heartbeat_at;
1819 let beaten = until(
1820 || {
1821 let seen = holder_of(&path).heartbeat_at;
1822 (seen > before).then_some(seen)
1823 },
1824 5_000,
1825 );
1826 assert!(beaten.is_some(), "the background heartbeat never advanced the lease");
1827 assert_eq!(holder_of(&path).owner, lock.owner());
1828 lock.release();
1829 }
1830}