lean_ctx/core/
path_locks.rs

1//! Process-wide per-path advisory locks.
2//!
3//! These in-process mutexes serialize concurrent operations on the *same* file
4//! path while letting operations on *different* paths run fully in parallel.
5//!
6//! Why this exists: tools like `ctx_read` and `ctx_edit` would otherwise contend
7//! on the single global cache write-lock for the entire duration of their disk
8//! I/O. When several agents (or sub-agents) hammer files concurrently, that
9//! global lock becomes a bottleneck and edits can time out waiting for it (see
10//! issue #320). A per-path lock keeps the contention scoped to the one file that
11//! actually needs serialization, so unrelated reads/edits never block each other.
12//!
13//! Lock ordering (see `LOCK_ORDERING.md`, L17): the inner registry mutex is held
14//! only long enough to clone the per-path `Arc<Mutex<()>>`, then released before
15//! the per-path lock itself is acquired. Never hold the registry mutex across the
16//! per-path lock, and never acquire a per-path lock while holding the global
17//! cache write-lock.
18
19use std::collections::HashMap;
20use std::sync::{Arc, Mutex};
21
22/// Upper bound on retained lock entries before we garbage-collect unused ones.
23const MAX_ENTRIES: usize = 500;
24
25/// Returns the shared advisory lock for `path`, creating it on first use.
26///
27/// The same path always yields the same `Arc<Mutex<()>>`, so callers across
28/// threads serialize on it. Different paths yield independent mutexes.
29pub fn per_file_lock(path: &str) -> Arc<Mutex<()>> {
30    static FILE_LOCKS: std::sync::OnceLock<Mutex<HashMap<String, Arc<Mutex<()>>>>> =
31        std::sync::OnceLock::new();
32    let map = FILE_LOCKS.get_or_init(|| Mutex::new(HashMap::new()));
33    let mut map = map.lock().unwrap_or_else(|poisoned| {
34        tracing::warn!("path_locks registry poisoned; recovering");
35        poisoned.into_inner()
36    });
37
38    // Bounded growth: drop entries no one else is holding a reference to. The
39    // `> 1` check keeps any lock that is currently in use by another caller.
40    if map.len() > MAX_ENTRIES {
41        map.retain(|_, v| Arc::strong_count(v) > 1);
42    }
43
44    map.entry(path.to_string())
45        .or_insert_with(|| Arc::new(Mutex::new(())))
46        .clone()
47}
48
49#[cfg(test)]
50mod tests {
51    use super::*;
52    use std::sync::Barrier;
53    use std::sync::atomic::{AtomicUsize, Ordering};
54
55    #[test]
56    fn same_path_returns_same_mutex() {
57        let a1 = per_file_lock("/tmp/path_locks_same.txt");
58        let a2 = per_file_lock("/tmp/path_locks_same.txt");
59        assert!(Arc::ptr_eq(&a1, &a2));
60    }
61
62    #[test]
63    fn different_paths_return_different_mutexes() {
64        let a = per_file_lock("/tmp/path_locks_a.txt");
65        let b = per_file_lock("/tmp/path_locks_b.txt");
66        assert!(!Arc::ptr_eq(&a, &b));
67    }
68
69    #[test]
70    fn serializes_concurrent_access_to_same_path() {
71        let counter = Arc::new(AtomicUsize::new(0));
72        let max_concurrent = Arc::new(AtomicUsize::new(0));
73        let barrier = Arc::new(Barrier::new(8));
74        let path = "/tmp/path_locks_serialize.txt";
75        let mut handles = Vec::new();
76        for _ in 0..8 {
77            let counter = Arc::clone(&counter);
78            let max_concurrent = Arc::clone(&max_concurrent);
79            let barrier = Arc::clone(&barrier);
80            handles.push(std::thread::spawn(move || {
81                barrier.wait();
82                let lock = per_file_lock(path);
83                let _guard = lock
84                    .lock()
85                    .unwrap_or_else(std::sync::PoisonError::into_inner);
86                let active = counter.fetch_add(1, Ordering::SeqCst) + 1;
87                max_concurrent.fetch_max(active, Ordering::SeqCst);
88                std::thread::sleep(std::time::Duration::from_millis(5));
89                counter.fetch_sub(1, Ordering::SeqCst);
90            }));
91        }
92        for h in handles {
93            h.join().unwrap();
94        }
95        assert_eq!(
96            max_concurrent.load(Ordering::SeqCst),
97            1,
98            "per-file lock must serialize same-path access"
99        );
100    }
101
102    #[test]
103    fn allows_parallel_access_to_different_paths() {
104        let counter = Arc::new(AtomicUsize::new(0));
105        let max_concurrent = Arc::new(AtomicUsize::new(0));
106        let barrier = Arc::new(Barrier::new(8));
107        let mut handles = Vec::new();
108        for i in 0..8 {
109            let counter = Arc::clone(&counter);
110            let max_concurrent = Arc::clone(&max_concurrent);
111            let barrier = Arc::clone(&barrier);
112            handles.push(std::thread::spawn(move || {
113                let path = format!("/tmp/path_locks_parallel_{i}.txt");
114                barrier.wait();
115                let lock = per_file_lock(&path);
116                let _guard = lock
117                    .lock()
118                    .unwrap_or_else(std::sync::PoisonError::into_inner);
119                let active = counter.fetch_add(1, Ordering::SeqCst) + 1;
120                max_concurrent.fetch_max(active, Ordering::SeqCst);
121                std::thread::sleep(std::time::Duration::from_millis(5));
122                counter.fetch_sub(1, Ordering::SeqCst);
123            }));
124        }
125        for h in handles {
126            h.join().unwrap();
127        }
128        assert!(
129            max_concurrent.load(Ordering::SeqCst) > 1,
130            "different paths must be allowed to run in parallel"
131        );
132    }
133}
lean_ctx/core/path_locks.rs

lean_ctx/core/
path_locks.rs
