1use std::collections::HashSet;
49use std::fs::{File, OpenOptions};
50use std::path::{Path, PathBuf};
51use std::sync::{Mutex, OnceLock};
52use std::time::{Duration, Instant};
53
54use crate::error::SqliteStoreError;
55
56fn held_exclusive_locks() -> &'static Mutex<HashSet<PathBuf>> {
65 static REGISTRY: OnceLock<Mutex<HashSet<PathBuf>>> = OnceLock::new();
66 REGISTRY.get_or_init(|| Mutex::new(HashSet::new()))
67}
68
69fn registry_key(lock_path: &Path) -> PathBuf {
70 std::fs::canonicalize(lock_path).unwrap_or_else(|_| lock_path.to_path_buf())
71}
72
73fn process_holds_exclusive(lock_path: &Path) -> bool {
74 held_exclusive_locks()
75 .lock()
76 .map(|held| held.contains(®istry_key(lock_path)))
77 .unwrap_or(false)
78}
79
80pub const FENCE_LOCK_SUFFIX: &str = "mfence";
82
83pub const PRESENCE_LOCK_SUFFIX: &str = "mpresence";
85
86pub fn fence_lock_path(db_path: &Path) -> PathBuf {
88 sibling_lock_path(db_path, FENCE_LOCK_SUFFIX)
89}
90
91pub fn presence_lock_path(db_path: &Path) -> PathBuf {
93 sibling_lock_path(db_path, PRESENCE_LOCK_SUFFIX)
94}
95
96fn sibling_lock_path(db_path: &Path, suffix: &str) -> PathBuf {
97 let mut name = db_path.file_name().unwrap_or_default().to_os_string();
98 name.push(".");
99 name.push(suffix);
100 db_path.with_file_name(name)
101}
102
103#[derive(Debug)]
110struct HeldLock(File);
111
112impl std::ops::Deref for HeldLock {
113 type Target = File;
114
115 fn deref(&self) -> &File {
116 &self.0
117 }
118}
119
120impl Drop for HeldLock {
121 fn drop(&mut self) {
122 let _ = self.0.unlock();
123 }
124}
125
126fn open_lock_file(lock_path: &Path) -> Option<File> {
127 OpenOptions::new()
130 .read(true)
131 .write(true)
132 .create(true)
133 .truncate(false)
134 .open(lock_path)
135 .ok()
136}
137
138#[derive(Debug)]
142pub struct OperationGuard {
143 _lock: Option<HeldLock>,
144}
145
146impl OperationGuard {
147 pub fn try_for_database_strict(db_path: &Path) -> Result<Self, SqliteStoreError> {
151 if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
152 return Err(SqliteStoreError::Io(std::io::Error::new(
153 std::io::ErrorKind::InvalidInput,
154 "strict operation custody requires an existing physical database",
155 )));
156 }
157 crate::profile::validate_database_file(db_path)?;
158 if !db_path.is_file() {
159 return Err(SqliteStoreError::Io(std::io::Error::new(
160 std::io::ErrorKind::NotFound,
161 "strict operation database is missing",
162 )));
163 }
164 #[cfg(not(any(target_os = "macos", target_os = "linux", windows)))]
165 {
166 Err(SqliteStoreError::Io(std::io::Error::new(
167 std::io::ErrorKind::Unsupported,
168 "strict operation custody requires descriptor-scoped locking",
169 )))
170 }
171 #[cfg(any(target_os = "macos", target_os = "linux", windows))]
172 {
173 let path = fence_lock_path(db_path);
174 crate::profile::validate_database_file(&path)?;
175 let file = OpenOptions::new()
176 .read(true)
177 .write(true)
178 .create(true)
179 .truncate(false)
180 .open(&path)?;
181 crate::profile::validate_database_file(&path)?;
182 if !file.metadata()?.is_file() {
183 return Err(SqliteStoreError::Io(std::io::Error::new(
184 std::io::ErrorKind::Unsupported,
185 "strict operation lock is not a regular file",
186 )));
187 }
188 match file.try_lock_shared() {
189 Ok(()) => Ok(Self {
190 _lock: Some(HeldLock(file)),
191 }),
192 Err(error) if is_would_block(&error) => {
193 Err(SqliteStoreError::MaintenanceFenceHeld {
194 path: db_path.to_path_buf(),
195 })
196 }
197 Err(error) => Err(SqliteStoreError::Io(std::io::Error::other(error))),
198 }
199 }
200 }
201
202 pub fn for_database(db_path: &Path) -> Result<Self, SqliteStoreError> {
208 if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
209 return Ok(Self { _lock: None });
210 }
211 crate::profile::validate_database_file(db_path)?;
213 let lock_path = fence_lock_path(db_path);
214 if process_holds_exclusive(&lock_path) {
215 return Ok(Self { _lock: None });
217 }
218 let Some(file) = open_lock_file(&lock_path) else {
219 return Ok(Self { _lock: None });
220 };
221 match file.try_lock_shared() {
222 Ok(()) => Ok(Self {
223 _lock: Some(HeldLock(file)),
224 }),
225 Err(err) if is_would_block(&err) => Err(SqliteStoreError::MaintenanceFenceHeld {
226 path: db_path.to_path_buf(),
227 }),
228 Err(_) => Ok(Self { _lock: None }),
230 }
231 }
232
233 pub fn admit_missing_database(db_path: &Path) -> Result<(), SqliteStoreError> {
242 if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
243 return Ok(());
244 }
245 let lock_path = fence_lock_path(db_path);
246 if process_holds_exclusive(&lock_path) {
247 return Ok(());
248 }
249 let Ok(file) = OpenOptions::new().read(true).write(true).open(&lock_path) else {
250 return Ok(());
251 };
252 match file.try_lock_shared() {
253 Err(err) if is_would_block(&err) => Err(SqliteStoreError::MaintenanceFenceHeld {
254 path: db_path.to_path_buf(),
255 }),
256 Ok(()) => {
258 drop(HeldLock(file));
259 Ok(())
260 }
261 Err(_) => Ok(()),
264 }
265 }
266}
267
268#[derive(Debug)]
280pub struct KeptConnectionPresence {
281 _lock: HeldLock,
282}
283
284impl KeptConnectionPresence {
285 pub fn acquire(db_path: &Path) -> Option<Self> {
303 if db_path.file_name().is_none() || db_path.as_os_str() == ":memory:" {
304 return None;
305 }
306 if process_holds_exclusive(&fence_lock_path(db_path)) {
307 return None;
308 }
309 let Ok(Some(_)) = crate::profile::validate_database_file(db_path) else {
312 return None;
313 };
314 let file = open_lock_file(&presence_lock_path(db_path))?;
315 file.try_lock_shared().ok()?;
316 Some(Self {
317 _lock: HeldLock(file),
318 })
319 }
320}
321
322#[derive(Debug)]
332pub struct ExclusiveFence {
333 _lock: HeldLock,
334 _presence: HeldLock,
335 lock_path: PathBuf,
336}
337
338impl ExclusiveFence {
339 pub fn try_acquire(db_path: &Path) -> Result<Option<Self>, SqliteStoreError> {
341 let lock_path = fence_lock_path(db_path);
342 let Some(fence) = try_lock_exclusive(&lock_path)? else {
343 return Ok(None);
344 };
345 let Some(presence) = try_lock_exclusive(&presence_lock_path(db_path))? else {
347 return Ok(None);
348 };
349 Ok(Some(Self::held(fence, presence, lock_path)))
350 }
351
352 pub fn acquire(db_path: &Path, deadline: Duration) -> Result<Self, SqliteStoreError> {
367 let started = Instant::now();
368 let lock_path = fence_lock_path(db_path);
369 let fence = loop {
370 if let Some(fence) = try_lock_exclusive(&lock_path)? {
371 break fence;
372 }
373 if started.elapsed() >= deadline {
374 return Err(SqliteStoreError::MaintenanceFenceHeld {
375 path: db_path.to_path_buf(),
376 });
377 }
378 std::thread::sleep(FENCE_POLL);
379 };
380 let presence_path = presence_lock_path(db_path);
381 loop {
382 if let Some(presence) = try_lock_exclusive(&presence_path)? {
383 return Ok(Self::held(fence, presence, lock_path));
384 }
385 if started.elapsed() >= deadline {
386 return Err(SqliteStoreError::MaintenanceFenceHeld {
388 path: db_path.to_path_buf(),
389 });
390 }
391 std::thread::sleep(FENCE_POLL);
392 }
393 }
394
395 fn held(fence: HeldLock, presence: HeldLock, lock_path: PathBuf) -> Self {
399 if let Ok(mut held) = held_exclusive_locks().lock() {
400 held.insert(registry_key(&lock_path));
401 }
402 Self {
403 _lock: fence,
404 _presence: presence,
405 lock_path,
406 }
407 }
408
409 pub fn lock_path(&self) -> &Path {
411 &self.lock_path
412 }
413}
414
415const FENCE_POLL: Duration = Duration::from_millis(25);
417
418fn try_lock_exclusive(lock_path: &Path) -> Result<Option<HeldLock>, SqliteStoreError> {
421 let file = OpenOptions::new()
422 .read(true)
423 .write(true)
424 .create(true)
425 .truncate(false)
426 .open(lock_path)?;
427 match file.try_lock() {
428 Ok(()) => Ok(Some(HeldLock(file))),
429 Err(err) if is_would_block(&err) => Ok(None),
430 Err(err) => Err(SqliteStoreError::Io(std::io::Error::other(err))),
431 }
432}
433
434impl Drop for ExclusiveFence {
435 fn drop(&mut self) {
439 if let Ok(mut held) = held_exclusive_locks().lock() {
440 held.remove(®istry_key(&self.lock_path));
441 }
442 }
443}
444
445fn is_would_block(err: &std::fs::TryLockError) -> bool {
446 matches!(err, std::fs::TryLockError::WouldBlock)
447}
448
449#[cfg(test)]
450#[allow(clippy::expect_used, clippy::unwrap_used, clippy::panic)]
451mod tests {
452 use super::*;
453
454 #[test]
455 fn shared_guards_coexist() {
456 let dir = tempfile::tempdir().expect("tempdir");
457 let db = dir.path().join("db.sqlite3");
458 let g1 = OperationGuard::for_database(&db).expect("g1");
459 let g2 = OperationGuard::for_database(&db).expect("g2");
460 drop((g1, g2));
461 }
462
463 #[test]
464 fn exclusive_fence_blocks_foreign_operations_typed() {
465 let dir = tempfile::tempdir().expect("tempdir");
466 let db = dir.path().join("db.sqlite3");
467 let lock_path = fence_lock_path(&db);
471 let foreign = OpenOptions::new()
472 .read(true)
473 .write(true)
474 .create(true)
475 .truncate(false)
476 .open(&lock_path)
477 .expect("open lock file");
478 foreign.try_lock().expect("foreign exclusive lock");
479 let err = OperationGuard::for_database(&db).expect_err("fence must block foreign ops");
480 assert!(matches!(err, SqliteStoreError::MaintenanceFenceHeld { .. }));
481 foreign.unlock().expect("release the foreign fence");
482 drop(foreign);
483 OperationGuard::for_database(&db).expect("released after drop");
484 }
485
486 #[test]
490 fn a_missing_database_honours_a_held_fence_and_creates_nothing() {
491 let dir = tempfile::tempdir().expect("tempdir");
492 let db = dir.path().join("db.sqlite3");
493 OperationGuard::admit_missing_database(&db).expect("no lock file: no fence");
494 assert!(
495 !fence_lock_path(&db).exists(),
496 "admission creates no lock file"
497 );
498
499 let lock_path = fence_lock_path(&db);
500 let foreign = OpenOptions::new()
501 .read(true)
502 .write(true)
503 .create(true)
504 .truncate(false)
505 .open(&lock_path)
506 .expect("open lock file");
507 foreign.try_lock().expect("foreign exclusive lock");
508 let err = OperationGuard::admit_missing_database(&db)
509 .expect_err("a foreign fence covers the missing database");
510 assert!(matches!(err, SqliteStoreError::MaintenanceFenceHeld { .. }));
511 foreign.unlock().expect("release the foreign fence");
512 drop(foreign);
513 OperationGuard::admit_missing_database(&db).expect("released after drop");
514
515 let own = ExclusiveFence::try_acquire(&db)
516 .expect("io")
517 .expect("own fence");
518 OperationGuard::admit_missing_database(&db).expect("the fence holder is admitted");
519 drop(own);
520 assert!(!db.exists(), "nothing created the database");
521 }
522
523 #[test]
524 fn exclusive_fence_waits_for_guard_drain() {
525 let dir = tempfile::tempdir().expect("tempdir");
526 let db = dir.path().join("db.sqlite3");
527 let guard = OperationGuard::for_database(&db).expect("guard");
528 assert!(
529 ExclusiveFence::try_acquire(&db).expect("io").is_none(),
530 "fence must not be acquirable while an operation is in flight"
531 );
532 let handle = std::thread::spawn(move || {
533 ExclusiveFence::acquire(&db, Duration::from_secs(5)).expect("acquire")
534 });
535 std::thread::sleep(Duration::from_millis(100));
536 drop(guard);
537 let fence = handle.join().expect("thread");
538 assert!(fence.lock_path().ends_with("db.sqlite3.mfence"));
539 }
540
541 #[test]
542 fn fence_holder_operations_self_admit() {
543 let dir = tempfile::tempdir().expect("tempdir");
544 let db = dir.path().join("db.sqlite3");
545 let fence = ExclusiveFence::try_acquire(&db)
546 .expect("io")
547 .expect("acquired");
548 let guard = OperationGuard::for_database(&db).expect("holder self-admission");
551 drop(guard);
552 drop(fence);
553 OperationGuard::for_database(&db).expect("released");
555 let fence = ExclusiveFence::try_acquire(&db)
557 .expect("io")
558 .expect("reacquired");
559 drop(fence);
560 }
561
562 #[test]
563 fn in_memory_paths_get_noop_guards() {
564 let guard = OperationGuard::for_database(Path::new(":memory:")).expect("noop");
565 drop(guard);
566 }
567
568 #[test]
569 fn fence_lock_path_shape() {
570 assert_eq!(
571 fence_lock_path(Path::new("/a/b/sessions.sqlite3")),
572 PathBuf::from("/a/b/sessions.sqlite3.mfence")
573 );
574 assert_eq!(
575 presence_lock_path(Path::new("/a/b/sessions.sqlite3")),
576 PathBuf::from("/a/b/sessions.sqlite3.mpresence")
577 );
578 }
579
580 fn existing_database(dir: &Path) -> PathBuf {
583 let db = dir.join("db.sqlite3");
584 std::fs::write(&db, b"").expect("database file");
585 db
586 }
587
588 #[test]
591 fn kept_presence_refuses_the_fence_and_a_refused_fence_holds_nothing() {
592 let dir = tempfile::tempdir().expect("tempdir");
593 let db = existing_database(dir.path());
594 let presence = KeptConnectionPresence::acquire(&db).expect("presence");
595 assert!(
596 ExclusiveFence::try_acquire(&db).expect("io").is_none(),
597 "the fence must not be taken while a connection is kept"
598 );
599 OperationGuard::for_database(&db).expect("a refused try holds no fence lock");
600 let err = ExclusiveFence::acquire(&db, Duration::from_millis(200))
601 .expect_err("a kept connection outlives the deadline");
602 assert!(matches!(err, SqliteStoreError::MaintenanceFenceHeld { .. }));
603 OperationGuard::for_database(&db).expect("an expired wait holds no fence lock");
604 drop(presence);
605 let fence = ExclusiveFence::try_acquire(&db)
606 .expect("io")
607 .expect("taken once the kept connection closed");
608 drop(fence);
609 }
610
611 #[test]
615 fn a_waiting_fence_refuses_the_kept_readers_next_read_and_completes_when_it_closes() {
616 let dir = tempfile::tempdir().expect("tempdir");
617 let db = existing_database(dir.path());
618 let presence = KeptConnectionPresence::acquire(&db).expect("presence");
619 let waiting = {
620 let db = db.clone();
621 std::thread::spawn(move || ExclusiveFence::acquire(&db, Duration::from_secs(10)))
622 };
623 let started = Instant::now();
625 loop {
626 match OperationGuard::for_database(&db) {
627 Err(SqliteStoreError::MaintenanceFenceHeld { .. }) => break,
628 Ok(guard) => drop(guard),
629 Err(other) => panic!("unexpected guard error: {other}"),
630 }
631 assert!(
632 started.elapsed() < Duration::from_secs(5),
633 "the waiting fence never took its fence lock"
634 );
635 std::thread::sleep(Duration::from_millis(5));
636 }
637 assert!(
638 !waiting.is_finished(),
639 "the fence waits for the kept reader"
640 );
641 drop(presence);
642 let fence = waiting
643 .join()
644 .expect("thread")
645 .expect("fence after the reader closed");
646 assert!(
647 KeptConnectionPresence::acquire(&db).is_none(),
648 "no connection is kept while the fence is held"
649 );
650 drop(fence);
651 KeptConnectionPresence::acquire(&db).expect("presence after release");
652 }
653
654 #[test]
657 fn presence_is_refused_while_the_fence_is_held() {
658 let dir = tempfile::tempdir().expect("tempdir");
659 let db = existing_database(dir.path());
660 let own = ExclusiveFence::try_acquire(&db)
661 .expect("io")
662 .expect("own fence");
663 assert!(KeptConnectionPresence::acquire(&db).is_none());
664 OperationGuard::for_database(&db).expect("holder self-admission");
666 drop(own);
667
668 let foreign = OpenOptions::new()
670 .read(true)
671 .write(true)
672 .create(true)
673 .truncate(false)
674 .open(presence_lock_path(&db))
675 .expect("open presence lock file");
676 foreign.try_lock().expect("foreign exclusive presence lock");
677 assert!(KeptConnectionPresence::acquire(&db).is_none());
678 foreign.unlock().expect("release the foreign presence lock");
679 drop(foreign);
680 KeptConnectionPresence::acquire(&db).expect("presence after release");
681 }
682
683 #[test]
685 fn presence_plants_nothing_next_to_a_refused_database() {
686 let dir = tempfile::tempdir().expect("tempdir");
687 let db = dir.path().join("db.sqlite3");
688 std::fs::create_dir(&db).expect("a directory where the database belongs");
689 assert!(KeptConnectionPresence::acquire(&db).is_none());
690 assert!(!presence_lock_path(&db).exists());
691 assert!(KeptConnectionPresence::acquire(Path::new(":memory:")).is_none());
692 }
693
694 #[test]
696 fn presence_plants_nothing_next_to_a_missing_database() {
697 let dir = tempfile::tempdir().expect("tempdir");
698 let db = dir.path().join("db.sqlite3");
699 assert!(KeptConnectionPresence::acquire(&db).is_none());
700 assert!(!presence_lock_path(&db).exists());
701 assert_eq!(
702 std::fs::read_dir(dir.path()).expect("read dir").count(),
703 0,
704 "nothing planted"
705 );
706 }
707
708 #[cfg(unix)]
713 struct DescriptorCopy(std::process::Child);
714
715 #[cfg(unix)]
716 impl DescriptorCopy {
717 fn spawn(file: &std::fs::File) -> Self {
719 let copy = file.try_clone().expect("a copy of the descriptor");
720 #[allow(clippy::disallowed_methods)] let child = std::process::Command::new("sleep")
722 .arg("600")
723 .stdin(copy)
724 .stdout(std::process::Stdio::null())
725 .stderr(std::process::Stdio::null())
726 .spawn()
727 .expect("a process holding the copy");
728 Self(child)
729 }
730 }
731
732 #[cfg(unix)]
733 impl Drop for DescriptorCopy {
734 fn drop(&mut self) {
735 if let Some(linger) = std::env::var("MEERKAT_TEST_DESCRIPTOR_COPY_LINGER_MS")
738 .ok()
739 .and_then(|millis| millis.parse::<u64>().ok())
740 {
741 std::thread::sleep(std::time::Duration::from_millis(linger));
742 }
743 let _ = self.0.kill();
744 let _ = self.0.wait();
745 }
746 }
747
748 #[cfg(unix)]
752 #[test]
753 fn a_released_fence_is_free_while_a_spawned_process_holds_its_descriptors() {
754 let dir = tempfile::tempdir().expect("tempdir");
755 let db = existing_database(dir.path());
756 let fence = ExclusiveFence::try_acquire(&db)
757 .expect("fence")
758 .expect("the fence is free");
759 let copies = [
760 DescriptorCopy::spawn(&fence._lock),
761 DescriptorCopy::spawn(&fence._presence),
762 ];
763 drop(fence);
764 OperationGuard::for_database(&db).expect("an operation passes a released fence");
765 assert!(
766 KeptConnectionPresence::acquire(&db).is_some(),
767 "a connection can be kept once the fence is released"
768 );
769 assert!(
770 ExclusiveFence::try_acquire(&db).expect("fence").is_some(),
771 "a released fence can be taken again"
772 );
773 drop(copies);
774 }
775
776 #[cfg(unix)]
779 #[test]
780 fn a_released_operation_guard_is_free_while_a_spawned_process_holds_its_descriptor() {
781 let dir = tempfile::tempdir().expect("tempdir");
782 let db = existing_database(dir.path());
783 let guard = OperationGuard::for_database(&db).expect("guard");
784 let copy = DescriptorCopy::spawn(guard._lock.as_ref().expect("a held guard"));
785 drop(guard);
786 assert!(
787 ExclusiveFence::try_acquire(&db).expect("fence").is_some(),
788 "the fence is free once the guard is released"
789 );
790 drop(copy);
791 }
792
793 #[cfg(unix)]
796 #[test]
797 fn a_released_presence_is_free_while_a_spawned_process_holds_its_descriptor() {
798 let dir = tempfile::tempdir().expect("tempdir");
799 let db = existing_database(dir.path());
800 let presence = KeptConnectionPresence::acquire(&db).expect("presence");
801 let copy = DescriptorCopy::spawn(&presence._lock);
802 drop(presence);
803 assert!(
804 ExclusiveFence::try_acquire(&db).expect("fence").is_some(),
805 "the fence is free once the presence is released"
806 );
807 drop(copy);
808 }
809}