stackless_core/
lockfile.rs

1//! Cross-process file locks keyed by path (ARCHITECTURE.md §2/§3).
2//!
3//! `create_new` with stale-holder detection by PID + start time — the
4//! same liveness identity op locks use. Daemon spawn, Stripe Projects,
5//! and git cache writers share this helper.
6
7use std::hash::{Hash, Hasher};
8use std::io::Write;
9use std::path::{Path, PathBuf};
10use std::time::{Duration, Instant};
11
12use crate::process::ProcessStamp;
13use crate::state::Store;
14use crate::types::{Pid, ProcessStartTime};
15
16const DEFAULT_POLL: Duration = Duration::from_millis(100);
17
18/// A held lock; released when dropped.
19#[derive(Debug)]
20pub struct FileLock {
21    path: PathBuf,
22}
23
24impl Drop for FileLock {
25    fn drop(&mut self) {
26        let _ = std::fs::remove_file(&self.path);
27    }
28}
29
30impl FileLock {
31    /// Try once; returns [`LockError::Held`] if a live holder has the lock.
32    pub fn try_acquire(path: &Path) -> Result<Self, LockError> {
33        if let Some(parent) = path.parent() {
34            std::fs::create_dir_all(parent).map_err(|err| LockError::CreateParent {
35                path: path.to_path_buf(),
36                detail: err.to_string(),
37            })?;
38        }
39        for _ in 0..2 {
40            match std::fs::OpenOptions::new()
41                .write(true)
42                .create_new(true)
43                .open(path)
44            {
45                Ok(mut file) => {
46                    let me = ProcessStamp::current();
47                    let _ = writeln!(file, "{} {}", me.pid.get(), me.start_time.get());
48                    return Ok(Self {
49                        path: path.to_path_buf(),
50                    });
51                }
52                Err(_) => {
53                    let stale = std::fs::read_to_string(path)
54                        .ok()
55                        .and_then(|content| {
56                            let mut parts = content.split_whitespace();
57                            let pid = parts.next()?.parse().ok()?;
58                            let start_time = parts.next()?.parse().ok()?;
59                            Some(ProcessStamp {
60                                pid: Pid::from_os(pid),
61                                start_time: ProcessStartTime::from_os(start_time),
62                            })
63                        })
64                        .is_none_or(|stamp| !stamp.is_alive());
65                    if stale {
66                        let _ = std::fs::remove_file(path);
67                        continue;
68                    }
69                    return Err(LockError::Held {
70                        path: path.to_path_buf(),
71                    });
72                }
73            }
74        }
75        Err(LockError::Held {
76            path: path.to_path_buf(),
77        })
78    }
79
80    /// Block until the lock is acquired, a stale holder is taken over, or
81    /// `budget` elapses.
82    pub fn acquire_with_wait(path: &Path, budget: Duration) -> Result<Self, LockError> {
83        let start = Instant::now();
84        loop {
85            match Self::try_acquire(path) {
86                Ok(lock) => return Ok(lock),
87                Err(LockError::Held { .. }) if start.elapsed() < budget => {
88                    std::thread::sleep(DEFAULT_POLL);
89                }
90                Err(err) => return Err(err),
91            }
92        }
93    }
94
95    /// Filesystem-safe digest of a path for lock file names.
96    pub fn path_key(path: &Path) -> String {
97        let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf());
98        let mut hasher = std::collections::hash_map::DefaultHasher::new();
99        canonical.hash(&mut hasher);
100        format!("{:016x}", hasher.finish())
101    }
102
103    /// `{state_dir}/locks/stripe/{digest}.lock` for a definition dir.
104    pub fn stripe_lock_path(definition_dir: &Path) -> PathBuf {
105        Store::state_dir()
106            .join("locks/stripe")
107            .join(format!("{}.lock", Self::path_key(definition_dir)))
108    }
109
110    /// `{state_dir}/locks/git-cache/{cache_key}.lock`.
111    pub fn git_cache_lock_path(cache_key: &str) -> PathBuf {
112        Store::state_dir()
113            .join("locks/git-cache")
114            .join(format!("{cache_key}.lock"))
115    }
116}
117
118#[derive(Debug, thiserror::Error)]
119pub enum LockError {
120    #[error("lock {path} is held by a live process")]
121    Held { path: PathBuf },
122    #[error("could not create lock parent for {path}: {detail}")]
123    CreateParent { path: PathBuf, detail: String },
124}
125
126#[cfg(test)]
127mod tests {
128    use super::*;
129    use std::sync::{Arc, Barrier};
130    use std::thread;
131
132    #[test]
133    fn second_holder_blocks_until_release() {
134        let dir = tempfile::tempdir().unwrap();
135        let path = dir.path().join("test.lock");
136
137        let lock = FileLock::try_acquire(&path).unwrap();
138        assert!(matches!(
139            FileLock::try_acquire(&path),
140            Err(LockError::Held { .. })
141        ));
142        drop(lock);
143        assert!(FileLock::try_acquire(&path).is_ok());
144    }
145
146    #[test]
147    fn concurrent_waiters_serialize() {
148        let dir = Arc::new(tempfile::tempdir().unwrap());
149        let n = 4;
150        let start = Arc::new(Barrier::new(n));
151        let mut handles = Vec::new();
152        for _ in 0..n {
153            let dir = Arc::clone(&dir);
154            let start = Arc::clone(&start);
155            handles.push(thread::spawn(move || {
156                start.wait();
157                let _lock = FileLock::acquire_with_wait(
158                    &dir.path().join("queue.lock"),
159                    Duration::from_secs(5),
160                )
161                .unwrap();
162            }));
163        }
164        for handle in handles {
165            handle.join().unwrap();
166        }
167    }
168}
stackless_core/lockfile.rs

stackless_core/
lockfile.rs
