1use std::fs;
26use std::io;
27use std::path::{Path, PathBuf};
28use std::sync::{Arc, Condvar, Mutex};
29use std::thread::{self, JoinHandle};
30use std::time::{Duration, Instant, SystemTime};
31
32use filetime::{FileTime, set_file_mtime};
33use same_file::Handle;
34use thiserror::Error;
35
36pub const DEFAULT_ATTEMPTS: u32 = 10;
38
39pub const DEFAULT_RETRY_DELAY: Duration = Duration::from_millis(20);
41
42pub const DEFAULT_STALE: Duration = Duration::from_secs(10);
46
47pub const DEFAULT_UPDATE: Duration = Duration::from_secs(5);
49
50#[derive(Clone, Debug, Eq, PartialEq)]
56pub struct LockOptions {
57 pub attempts: u32,
59 pub retry_delay: Duration,
61 pub stale: Duration,
63 pub update: Duration,
68 pub lockfile_path: Option<PathBuf>,
74}
75
76impl Default for LockOptions {
77 fn default() -> Self {
78 Self {
79 attempts: DEFAULT_ATTEMPTS,
80 retry_delay: DEFAULT_RETRY_DELAY,
81 stale: DEFAULT_STALE,
82 update: DEFAULT_UPDATE,
83 lockfile_path: None,
84 }
85 }
86}
87
88impl LockOptions {
89 #[must_use]
91 pub fn new() -> Self {
92 Self::default()
93 }
94
95 #[must_use]
97 pub fn attempts(mut self, attempts: u32) -> Self {
98 self.attempts = attempts.max(1);
99 self
100 }
101
102 #[must_use]
104 pub fn retry_delay(mut self, retry_delay: Duration) -> Self {
105 self.retry_delay = retry_delay;
106 self
107 }
108
109 #[must_use]
111 pub fn stale(mut self, stale: Duration) -> Self {
112 self.stale = stale;
113 self
114 }
115
116 #[must_use]
118 pub fn update(mut self, update: Duration) -> Self {
119 self.update = update;
120 self
121 }
122
123 #[must_use]
125 pub fn lockfile_path(mut self, lockfile_path: impl Into<PathBuf>) -> Self {
126 self.lockfile_path = Some(lockfile_path.into());
127 self
128 }
129}
130
131#[derive(Debug, Error)]
133pub enum LockError {
134 #[error("Lock file is already being held")]
136 Contended {
137 target: PathBuf,
139 lock_path: PathBuf,
141 },
142 #[error("lock I/O failed for {}: {source}", lock_path.display())]
144 Io {
145 target: PathBuf,
147 lock_path: PathBuf,
149 #[source]
151 source: io::Error,
152 },
153}
154
155impl LockError {
156 #[must_use]
158 pub const fn is_contended(&self) -> bool {
159 matches!(self, Self::Contended { .. })
160 }
161
162 #[must_use]
164 pub fn code(&self) -> &'static str {
165 match self {
166 Self::Contended { .. } => "ELOCKED",
167 Self::Io { .. } => "EIO",
168 }
169 }
170}
171
172#[derive(Debug)]
174struct StopSignal {
175 stopped: Mutex<bool>,
176 cvar: Condvar,
177}
178
179impl StopSignal {
180 fn new() -> Self {
181 Self {
182 stopped: Mutex::new(false),
183 cvar: Condvar::new(),
184 }
185 }
186
187 fn stop(&self) {
189 if let Ok(mut guard) = self.stopped.lock() {
190 *guard = true;
191 self.cvar.notify_all();
192 }
193 }
194
195 fn wait_timeout(&self, total: Duration) -> bool {
197 let Ok(mut guard) = self.stopped.lock() else {
198 return true;
200 };
201 if *guard {
202 return true;
203 }
204 let deadline = Instant::now() + total;
205 loop {
206 if *guard {
207 return true;
208 }
209 let now = Instant::now();
210 if now >= deadline {
211 return false;
212 }
213 let remaining = deadline.saturating_duration_since(now);
214 let Ok((next, wait_result)) = self.cvar.wait_timeout(guard, remaining) else {
215 return true;
216 };
217 guard = next;
218 if *guard {
219 return true;
220 }
221 if wait_result.timed_out() {
222 return false;
223 }
224 }
225 }
226}
227
228#[derive(Debug)]
234pub struct LockGuard {
235 target: PathBuf,
236 lock_path: PathBuf,
237 identity: Arc<Handle>,
239 stop: Arc<StopSignal>,
240 refresh_thread: Option<JoinHandle<()>>,
241 released: bool,
242}
243
244impl LockGuard {
245 pub fn acquire(target: impl AsRef<Path>) -> Result<Self, LockError> {
253 Self::acquire_with(target, &LockOptions::default())
254 }
255
256 pub fn acquire_with(
264 target: impl AsRef<Path>,
265 options: &LockOptions,
266 ) -> Result<Self, LockError> {
267 let target_ref = target.as_ref();
268 let target = absolute_path(target_ref);
269 let lock_path = match &options.lockfile_path {
270 Some(custom) => absolute_path(custom),
271 None => default_lock_path(&target),
272 };
273
274 let attempts = options.attempts.max(1);
275 let mut last_contended = false;
276
277 for attempt in 1..=attempts {
278 match try_create_lock(&lock_path, options.stale, true) {
279 Ok(()) => {
280 let identity = match Handle::from_path(&lock_path) {
281 Ok(handle) => Arc::new(handle),
282 Err(source) => {
283 let _ = fs::remove_dir(&lock_path);
286 return Err(LockError::Io {
287 target,
288 lock_path,
289 source,
290 });
291 }
292 };
293 return Ok(Self::from_acquired(
294 target,
295 lock_path,
296 identity,
297 options.update,
298 ));
299 }
300 Err(OnceError::Contended) => {
301 last_contended = true;
302 if attempt == attempts {
303 break;
304 }
305 if !options.retry_delay.is_zero() {
306 thread::sleep(options.retry_delay);
307 }
308 }
309 Err(OnceError::Io(source)) => {
310 return Err(LockError::Io {
311 target,
312 lock_path,
313 source,
314 });
315 }
316 }
317 }
318
319 if last_contended {
320 Err(LockError::Contended { target, lock_path })
321 } else {
322 Err(LockError::Io {
323 target,
324 lock_path,
325 source: io::Error::other("lock acquisition produced no result"),
326 })
327 }
328 }
329
330 #[must_use]
332 pub fn target(&self) -> &Path {
333 &self.target
334 }
335
336 #[must_use]
338 pub fn lock_path(&self) -> &Path {
339 &self.lock_path
340 }
341
342 pub fn release(mut self) -> Result<(), LockError> {
350 self.release_inner()
351 }
352
353 fn from_acquired(
354 target: PathBuf,
355 lock_path: PathBuf,
356 identity: Arc<Handle>,
357 update: Duration,
358 ) -> Self {
359 let stop = Arc::new(StopSignal::new());
360 let refresh_thread = start_refresh_thread(
361 lock_path.clone(),
362 Arc::clone(&identity),
363 update,
364 Arc::clone(&stop),
365 );
366 Self {
367 target,
368 lock_path,
369 identity,
370 stop,
371 refresh_thread,
372 released: false,
373 }
374 }
375
376 fn release_inner(&mut self) -> Result<(), LockError> {
377 if self.released {
378 return Ok(());
379 }
380 self.released = true;
381 self.stop.stop();
382 if let Some(handle) = self.refresh_thread.take() {
383 let _ = handle.join();
385 }
386
387 match still_our_lock(&self.lock_path, &self.identity) {
391 Ok(true) => match fs::remove_dir(&self.lock_path) {
392 Ok(()) => Ok(()),
393 Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(()),
394 Err(source) => Err(LockError::Io {
395 target: self.target.clone(),
396 lock_path: self.lock_path.clone(),
397 source,
398 }),
399 },
400 Ok(false) | Err(_) => Ok(()),
402 }
403 }
404}
405
406impl Drop for LockGuard {
407 fn drop(&mut self) {
408 let _ = self.release_inner();
409 }
410}
411
412enum OnceError {
414 Contended,
415 Io(io::Error),
416}
417
418fn try_create_lock(
419 lock_path: &Path,
420 stale: Duration,
421 allow_stale_reclaim: bool,
422) -> Result<(), OnceError> {
423 match fs::create_dir(lock_path) {
424 Ok(()) => Ok(()),
425 Err(err) if is_already_exists(&err) => {
426 if !allow_stale_reclaim || stale.is_zero() {
427 return Err(OnceError::Contended);
428 }
429 match fs::metadata(lock_path) {
430 Err(meta_err) if meta_err.kind() == io::ErrorKind::NotFound => {
431 try_create_lock(lock_path, Duration::ZERO, false)
433 }
434 Err(meta_err) => Err(OnceError::Io(meta_err)),
435 Ok(meta) => {
436 if !is_lock_stale(&meta, stale) {
437 return Err(OnceError::Contended);
438 }
439 match fs::remove_dir(lock_path) {
440 Ok(()) => {}
441 Err(remove_err) if remove_err.kind() == io::ErrorKind::NotFound => {}
442 Err(remove_err) => return Err(OnceError::Io(remove_err)),
443 }
444 try_create_lock(lock_path, Duration::ZERO, false)
447 }
448 }
449 }
450 Err(err) => Err(OnceError::Io(err)),
451 }
452}
453
454fn is_already_exists(err: &io::Error) -> bool {
455 err.kind() == io::ErrorKind::AlreadyExists
456}
457
458fn is_lock_stale(meta: &fs::Metadata, stale: Duration) -> bool {
459 let Ok(mtime) = meta.modified() else {
460 return false;
462 };
463 match SystemTime::now().duration_since(mtime) {
464 Ok(age) => age > stale,
466 Err(_) => false,
468 }
469}
470
471fn default_lock_path(target: &Path) -> PathBuf {
472 let mut os = target.as_os_str().to_os_string();
473 os.push(".lock");
474 PathBuf::from(os)
475}
476
477fn absolute_path(path: &Path) -> PathBuf {
478 match std::path::absolute(path) {
482 Ok(abs) => abs,
483 Err(_) => {
484 if path.is_absolute() {
485 path.to_path_buf()
486 } else {
487 std::env::current_dir().map_or_else(|_| path.to_path_buf(), |cwd| cwd.join(path))
488 }
489 }
490 }
491}
492
493fn still_our_lock(lock_path: &Path, identity: &Handle) -> io::Result<bool> {
495 match Handle::from_path(lock_path) {
496 Ok(current) => Ok(current == *identity),
497 Err(err) if err.kind() == io::ErrorKind::NotFound => Ok(false),
498 Err(err) => Err(err),
499 }
500}
501
502fn start_refresh_thread(
503 lock_path: PathBuf,
504 identity: Arc<Handle>,
505 update: Duration,
506 stop: Arc<StopSignal>,
507) -> Option<JoinHandle<()>> {
508 if update.is_zero() {
509 return None;
510 }
511
512 Some(thread::spawn(move || {
513 loop {
514 if stop.wait_timeout(update) {
515 break;
516 }
517 match still_our_lock(&lock_path, &identity) {
521 Ok(true) => {
522 let _ = set_file_mtime(&lock_path, FileTime::now());
523 }
524 Ok(false) | Err(_) => break,
525 }
526 }
527 }))
528}
529
530#[cfg(test)]
531mod tests {
532 use super::*;
533 use std::error::Error;
534 use std::sync::mpsc;
535 use std::time::Duration;
536
537 type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
538
539 #[derive(Debug)]
540 struct TestFailure(String);
541
542 impl std::fmt::Display for TestFailure {
543 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
544 f.write_str(&self.0)
545 }
546 }
547
548 impl Error for TestFailure {}
549
550 fn fail(message: impl Into<String>) -> Box<dyn Error + Send + Sync> {
551 Box::new(TestFailure(message.into()))
552 }
553
554 fn make_temp_dir() -> Result<PathBuf, Box<dyn Error + Send + Sync>> {
555 let nanos = SystemTime::now()
556 .duration_since(SystemTime::UNIX_EPOCH)
557 .map_or(0, |d| d.as_nanos());
558 let base =
559 std::env::temp_dir().join(format!("pi-lockfile-{}-{}", std::process::id(), nanos));
560 fs::create_dir_all(&base)?;
561 Ok(base)
562 }
563
564 fn short_options() -> LockOptions {
565 LockOptions::new()
566 .attempts(3)
567 .retry_delay(Duration::from_millis(5))
568 .stale(Duration::from_millis(40))
569 .update(Duration::from_millis(10))
570 }
571
572 fn require_contended(
573 result: Result<LockGuard, LockError>,
574 ) -> Result<LockError, Box<dyn Error + Send + Sync>> {
575 match result {
576 Ok(_) => Err(fail("expected contention, acquired lock")),
577 Err(err) if err.is_contended() => Ok(err),
578 Err(err) => Err(fail(format!("expected contention, got {err}"))),
579 }
580 }
581
582 fn require_io(
583 result: Result<LockGuard, LockError>,
584 ) -> Result<LockError, Box<dyn Error + Send + Sync>> {
585 match result {
586 Ok(_) => Err(fail("expected I/O error, acquired lock")),
587 Err(err) if !err.is_contended() => Ok(err),
588 Err(err) => Err(fail(format!("expected I/O error, got {err}"))),
589 }
590 }
591
592 #[test]
593 fn acquires_when_target_does_not_exist() -> TestResult {
594 let dir = make_temp_dir()?;
595 let target = dir.join("settings.json");
596 if target.exists() {
597 return Err(fail("target should not exist before acquire"));
598 }
599
600 let guard = LockGuard::acquire(&target)?;
601 if !guard.lock_path().exists() {
602 return Err(fail("lock directory missing after acquire"));
603 }
604 if !guard.lock_path().is_dir() {
605 return Err(fail("lock path is not a directory"));
606 }
607 let ends_with_lock = guard
608 .lock_path()
609 .file_name()
610 .and_then(|n| n.to_str())
611 .is_some_and(|n| n.ends_with("settings.json.lock"));
612 if !ends_with_lock {
613 return Err(fail("lock path suffix mismatch"));
614 }
615 drop(guard);
616 if default_lock_path(&absolute_path(&target)).exists() {
617 return Err(fail("lock directory remained after drop"));
618 }
619 let _ = fs::remove_dir_all(&dir);
620 Ok(())
621 }
622
623 #[test]
624 fn release_on_drop_removes_only_owned_lock_dir() -> TestResult {
625 let dir = make_temp_dir()?;
626 let target = dir.join("trust.json");
627 let foreign = dir.join("other.lock");
628 fs::create_dir(&foreign)?;
629
630 {
631 let guard = LockGuard::acquire(&target)?;
632 if !guard.lock_path().exists() {
633 return Err(fail("owned lock missing"));
634 }
635 if !foreign.exists() {
636 return Err(fail("foreign lock missing during hold"));
637 }
638 }
639
640 if default_lock_path(&absolute_path(&target)).exists() {
641 return Err(fail("owned lock remained after drop"));
642 }
643 if !foreign.exists() {
644 return Err(fail("unrelated lock dir must remain"));
645 }
646 let _ = fs::remove_dir_all(&dir);
647 Ok(())
648 }
649
650 #[test]
651 fn mutual_exclusion_second_holder_is_contended() -> TestResult {
652 let dir = make_temp_dir()?;
653 let target = dir.join("settings.json");
654 let first = LockGuard::acquire(&target)?;
655
656 let target_for_thread = target.clone();
657 let (tx, rx) = mpsc::channel();
658 let handle = thread::spawn(move || {
659 let result = LockGuard::acquire_with(
660 &target_for_thread,
661 &LockOptions::new()
662 .attempts(2)
663 .retry_delay(Duration::from_millis(5)),
664 );
665 let is_err = result.is_err();
666 let _ = tx.send(is_err);
667 result
668 });
669
670 let contended = match rx.recv_timeout(Duration::from_millis(200)) {
671 Ok(value) => value,
672 Err(err) => return Err(fail(format!("join signal: {err}"))),
673 };
674 if !contended {
675 return Err(fail("second acquire must fail while first holds"));
676 }
677
678 let Ok(thread_result) = handle.join() else {
679 return Err(fail("second-holder thread panicked"));
680 };
681 let err = require_contended(thread_result)?;
682 if err.code() != "ELOCKED" {
683 return Err(fail(format!("expected ELOCKED, got {}", err.code())));
684 }
685
686 drop(first);
687 let second = LockGuard::acquire(&target)?;
688 drop(second);
689 let _ = fs::remove_dir_all(&dir);
690 Ok(())
691 }
692
693 #[test]
694 fn stale_lock_is_reclaimed() -> TestResult {
695 let dir = make_temp_dir()?;
696 let target = dir.join("settings.json");
697 let lock_path = default_lock_path(&absolute_path(&target));
698 fs::create_dir(&lock_path)?;
699
700 let past = SystemTime::now()
701 .checked_sub(Duration::from_secs(30))
702 .ok_or_else(|| fail("past time underflow"))?;
703 set_file_mtime(&lock_path, FileTime::from_system_time(past))?;
704
705 let guard = LockGuard::acquire_with(
706 &target,
707 &LockOptions::new()
708 .attempts(2)
709 .retry_delay(Duration::ZERO)
710 .stale(Duration::from_millis(50))
711 .update(Duration::from_millis(20)),
712 )?;
713 if !guard.lock_path().exists() {
714 return Err(fail("reclaimed lock missing"));
715 }
716 drop(guard);
717 let _ = fs::remove_dir_all(&dir);
718 Ok(())
719 }
720
721 #[test]
722 fn fresh_foreign_lock_times_out_as_contended() -> TestResult {
723 let dir = make_temp_dir()?;
724 let target = dir.join("settings.json");
725 let lock_path = default_lock_path(&absolute_path(&target));
726 fs::create_dir(&lock_path)?;
727
728 let started = Instant::now();
729 let err = require_contended(LockGuard::acquire_with(
730 &target,
731 &LockOptions::new()
732 .attempts(3)
733 .retry_delay(Duration::from_millis(5))
734 .stale(Duration::from_secs(30))
735 .update(Duration::from_secs(15)),
736 ))?;
737 let elapsed = started.elapsed();
738
739 if err.code() != "ELOCKED" {
740 return Err(fail(format!("expected ELOCKED, got {}", err.code())));
741 }
742 if elapsed > Duration::from_millis(200) {
743 return Err(fail(format!(
744 "retries must stay within 10*20ms production bound, elapsed={elapsed:?}"
745 )));
746 }
747 if !lock_path.exists() {
748 return Err(fail("foreign lock must remain"));
749 }
750 let _ = fs::remove_dir_all(&dir);
751 Ok(())
752 }
753
754 #[test]
755 fn held_lock_stays_fresh_past_stale_threshold() -> TestResult {
756 let dir = make_temp_dir()?;
757 let target = dir.join("settings.json");
758 let opts = short_options();
759
760 let guard = LockGuard::acquire_with(&target, &opts)?;
761
762 thread::sleep(Duration::from_millis(100));
765
766 let err = require_contended(LockGuard::acquire_with(
767 &target,
768 &LockOptions::new()
769 .attempts(1)
770 .retry_delay(Duration::ZERO)
771 .stale(opts.stale)
772 .update(opts.update),
773 ))?;
774 if !err.is_contended() {
775 return Err(fail("live holder must not be stolen after stale window"));
776 }
777
778 drop(guard);
779 let reacquired = LockGuard::acquire_with(&target, &opts)?;
780 drop(reacquired);
781 let _ = fs::remove_dir_all(&dir);
782 Ok(())
783 }
784
785 #[test]
786 fn unrelated_io_errors_propagate_without_retry_delay() -> TestResult {
787 let dir = make_temp_dir()?;
788 let target = dir.join("missing").join("nested").join("settings.json");
792 let started = Instant::now();
793 let err = require_io(LockGuard::acquire_with(
794 &target,
795 &LockOptions::new()
796 .attempts(10)
797 .retry_delay(Duration::from_millis(20)),
798 ))?;
799 let elapsed = started.elapsed();
800
801 if err.is_contended() {
802 return Err(fail("non-contention path returned contended"));
803 }
804 if err.code() != "EIO" {
805 return Err(fail(format!("expected EIO, got {}", err.code())));
806 }
807 if elapsed >= Duration::from_millis(50) {
808 return Err(fail(format!(
809 "non-contention errors must not sleep through the retry budget, elapsed={elapsed:?}"
810 )));
811 }
812 let _ = fs::remove_dir_all(&dir);
813 Ok(())
814 }
815
816 #[test]
817 fn custom_lockfile_path_is_honored() -> TestResult {
818 let dir = make_temp_dir()?;
819 let target = dir.join("store");
820 fs::create_dir_all(&target)?;
821 let custom = dir.join("trust.json.lock");
822
823 let guard =
824 LockGuard::acquire_with(&target, &LockOptions::new().lockfile_path(custom.clone()))?;
825 if guard.lock_path() != absolute_path(&custom).as_path() {
826 return Err(fail("custom lock path not honored"));
827 }
828 if !(custom.exists() || absolute_path(&custom).exists()) {
829 return Err(fail("custom lock directory missing while held"));
830 }
831 drop(guard);
832 if absolute_path(&custom).exists() {
833 return Err(fail("custom lock directory remained after drop"));
834 }
835 let _ = fs::remove_dir_all(&dir);
836 Ok(())
837 }
838
839 #[test]
840 fn explicit_release_is_idempotent_with_drop() -> TestResult {
841 let dir = make_temp_dir()?;
842 let target = dir.join("settings.json");
843 let guard = LockGuard::acquire(&target)?;
844 let lock_path = guard.lock_path().to_path_buf();
845 guard.release()?;
846 if lock_path.exists() {
847 return Err(fail("lock remained after explicit release"));
848 }
849 let _ = fs::remove_dir_all(&dir);
850 Ok(())
851 }
852
853 #[test]
854 fn replaced_lock_dir_is_neither_refreshed_nor_deleted_by_old_guard() -> TestResult {
855 let dir = make_temp_dir()?;
856 let target = dir.join("settings.json");
857 let lock_path = default_lock_path(&absolute_path(&target));
858
859 let guard = LockGuard::acquire_with(
862 &target,
863 &LockOptions::new()
864 .attempts(1)
865 .retry_delay(Duration::ZERO)
866 .stale(Duration::from_millis(30))
867 .update(Duration::from_millis(200)),
868 )?;
869
870 let original_identity = Handle::from_path(&lock_path)?;
871
872 fs::remove_dir(&lock_path)?;
875 fs::create_dir(&lock_path)?;
876 let replacement_identity = Handle::from_path(&lock_path)?;
877 if original_identity == replacement_identity {
878 return Err(fail("replacement must be a distinct filesystem object"));
879 }
880
881 let past = SystemTime::now()
884 .checked_sub(Duration::from_mins(1))
885 .ok_or_else(|| fail("past time underflow"))?;
886 set_file_mtime(&lock_path, FileTime::from_system_time(past))?;
887 let aged_mtime = fs::metadata(&lock_path).and_then(|m| m.modified())?;
888
889 thread::sleep(Duration::from_millis(250));
892
893 let after_wait_mtime = fs::metadata(&lock_path).and_then(|m| m.modified())?;
894 if aged_mtime != after_wait_mtime {
895 return Err(fail("old guard must not refresh a replaced lock directory"));
896 }
897 if Handle::from_path(&lock_path)? != replacement_identity {
898 return Err(fail("replacement identity changed during wait"));
899 }
900
901 drop(guard);
903 if !lock_path.exists() {
904 return Err(fail("old guard must not remove a replaced lock directory"));
905 }
906 if Handle::from_path(&lock_path)? != replacement_identity {
907 return Err(fail("replacement identity lost after old guard drop"));
908 }
909
910 let _ = fs::remove_dir_all(&dir);
911 Ok(())
912 }
913}