xianyun_deploy/registrycache/
lock.rs

1//! Cross-process serialization for registry-cache state mutations.
2//!
3//! Ported from `internal/registrycache/lock.go`.
4
5use std::fs::{self, OpenOptions};
6use std::os::fd::AsRawFd;
7
8use thiserror::Error;
9
10use super::paths::{self, PathError};
11
12/// The basename of the flock used to serialize all state-mutating
13/// operations (`ensure`, `ensure_stopped_for_profile`).
14const LOCK_FILE: &str = ".lock";
15
16/// Errors from acquiring the registry-cache flock.
17#[derive(Debug, Error)]
18pub enum LockError {
19    /// Failed resolving the registry-cache directory.
20    #[error(transparent)]
21    Path(#[from] PathError),
22    /// Failed creating the registry-cache directory.
23    #[error("creating registry cache dir: {0}")]
24    CreateDir(std::io::Error),
25    /// Failed opening (or creating) the lock file.
26    #[error("opening lock file: {0}")]
27    OpenLockFile(std::io::Error),
28    /// The `flock(2)` syscall itself failed.
29    #[error("acquiring flock: {0}")]
30    Flock(std::io::Error),
31}
32
33/// Acquires an exclusive flock on the registry-cache lock file for the
34/// duration of `f`. The lock file is created on first use. Released
35/// when `f` returns (including via panic-unwind, since the underlying
36/// file descriptor is closed by `Drop`).
37///
38/// Use this around every operation that touches refs, the PID file,
39/// the config file, or the zot process state. Generic over the
40/// caller's error type `E` (which must be constructible from
41/// [`LockError`]) so callers can propagate their own error variants
42/// from `f` without a boxed catch-all.
43pub fn with_lock<T, E, F>(f: F) -> Result<T, E>
44where
45    F: FnOnce() -> Result<T, E>,
46    E: From<LockError>,
47{
48    let dir = paths::global_dir().map_err(LockError::from)?;
49
50    fs::create_dir_all(&dir).map_err(LockError::CreateDir)?;
51
52    let file = OpenOptions::new()
53        .create(true)
54        .truncate(false)
55        .read(true)
56        .write(true)
57        .open(dir.join(LOCK_FILE))
58        .map_err(LockError::OpenLockFile)?;
59
60    // SAFETY: `file` stays alive (and thus its fd valid) for the
61    // duration of the flock call; flock(2) is a well-defined syscall
62    // given a valid fd.
63    let rc = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) };
64    if rc != 0 {
65        return Err(LockError::Flock(std::io::Error::last_os_error()).into());
66    }
67
68    let result = f();
69
70    // Best-effort unlock; the fd closing on drop would release the lock
71    // anyway, but do it explicitly (and ignore errors) to mirror the Go
72    // `defer` semantics exactly.
73    // SAFETY: same fd, still valid.
74    let _ = unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_UN) };
75
76    result
77}
78
79#[cfg(test)]
80mod tests {
81    use std::sync::atomic::{AtomicI64, Ordering};
82    use std::sync::{Arc, Mutex};
83    use std::thread;
84    use std::time::Duration;
85
86    use super::*;
87    use crate::registrycache::test_support::with_home;
88
89    // Ported from internal/registrycache/lock_test.go:12 (TestWithLockSerializesCalls)
90    #[test]
91    fn test_with_lock_serializes_calls() {
92        with_home(|_home| {
93            const N: usize = 5;
94
95            let concurrent = Arc::new(AtomicI64::new(0));
96            let max_seen = Arc::new(AtomicI64::new(0));
97            let errs = Arc::new(Mutex::new(Vec::new()));
98
99            let handles: Vec<_> = (0..N)
100                .map(|_| {
101                    let concurrent = Arc::clone(&concurrent);
102                    let max_seen = Arc::clone(&max_seen);
103                    let errs = Arc::clone(&errs);
104
105                    thread::spawn(move || {
106                        let result: Result<(), LockError> = with_lock(|| {
107                            let now = concurrent.fetch_add(1, Ordering::SeqCst) + 1;
108
109                            loop {
110                                let old_max = max_seen.load(Ordering::SeqCst);
111                                if now <= old_max
112                                    || max_seen
113                                        .compare_exchange(
114                                            old_max,
115                                            now,
116                                            Ordering::SeqCst,
117                                            Ordering::SeqCst,
118                                        )
119                                        .is_ok()
120                                {
121                                    break;
122                                }
123                            }
124
125                            thread::sleep(Duration::from_millis(50));
126                            concurrent.fetch_sub(1, Ordering::SeqCst);
127
128                            Ok(())
129                        });
130
131                        if let Err(err) = result {
132                            errs.lock().unwrap().push(err.to_string());
133                        }
134                    })
135                })
136                .collect();
137
138            for h in handles {
139                h.join().unwrap();
140            }
141
142            let errs = errs.lock().unwrap();
143            assert!(errs.is_empty(), "unexpected errors: {errs:?}");
144
145            // Critical section must have been single-threaded.
146            assert_eq!(max_seen.load(Ordering::SeqCst), 1);
147        });
148    }
149}
xianyun_deploy/registrycache/lock.rs

xianyun_deploy/registrycache/
lock.rs
