zeppelin_embed/lifecycle/
lock.rs

1//! Single-writer ownership via a process-local registry and `fcntl` lock.
2//!
3//! POSIX record locks are not inherited across `fork`, but same-process locks
4//! do not conflict and closing any descriptor for the lock file releases them.
5//! The registry preserves same-process exclusion. Nothing else in this process
6//! may open [`STORE_LOCK_FILE`].
7
8#[cfg(unix)]
9use std::collections::BTreeSet;
10use std::fs::{File, OpenOptions};
11#[cfg(unix)]
12use std::os::fd::AsRawFd;
13#[cfg(unix)]
14use std::os::unix::fs::MetadataExt as _;
15use std::path::{Path, PathBuf};
16#[cfg(unix)]
17use std::sync::{Mutex, MutexGuard};
18
19/// Persistent lock filename; only [`StoreLock`] may open this file.
20pub const STORE_LOCK_FILE: &str = "writer.lock";
21
22#[cfg(unix)]
23type StoreKey = (u64, u64);
24
25#[cfg(unix)]
26static HELD_STORES: Mutex<BTreeSet<StoreKey>> = Mutex::new(BTreeSet::new());
27
28/// Exclusive writer ownership automatically released when the process dies.
29#[derive(Debug)]
30pub struct StoreLock {
31    file: Option<File>,
32    #[cfg(unix)]
33    registry_key: StoreKey,
34}
35
36impl StoreLock {
37    /// Attempts to acquire exclusive non-blocking ownership for one store.
38    pub fn acquire(directory: &Path) -> Result<Self, StoreLockError> {
39        let path = directory.join(STORE_LOCK_FILE);
40        #[cfg(unix)]
41        let registry_key = {
42            let metadata = std::fs::metadata(directory).map_err(|source| StoreLockError::Io {
43                path: directory.to_path_buf(),
44                source,
45            })?;
46            let key = (metadata.dev(), metadata.ino());
47            if !held_stores().insert(key) {
48                return Err(StoreLockError::Io {
49                    path,
50                    source: std::io::Error::from(std::io::ErrorKind::WouldBlock),
51                });
52            }
53            key
54        };
55        let file = OpenOptions::new()
56            .create(true)
57            .truncate(false)
58            .read(true)
59            .write(true)
60            .open(&path)
61            .map_err(|source| {
62                #[cfg(unix)]
63                held_stores().remove(&registry_key);
64                StoreLockError::Io {
65                    path: path.clone(),
66                    source,
67                }
68            })?;
69        if let Err(source) = lock_exclusive(&file) {
70            drop(file);
71            #[cfg(unix)]
72            held_stores().remove(&registry_key);
73            return Err(StoreLockError::Io { path, source });
74        }
75        Ok(Self {
76            file: Some(file),
77            #[cfg(unix)]
78            registry_key,
79        })
80    }
81}
82
83#[cfg(unix)]
84fn held_stores() -> MutexGuard<'static, BTreeSet<StoreKey>> {
85    match HELD_STORES.lock() {
86        Ok(held) => held,
87        Err(poisoned) => poisoned.into_inner(),
88    }
89}
90
91#[cfg(unix)]
92fn lock_exclusive(file: &File) -> std::io::Result<()> {
93    let mut lock = unsafe {
94        // SAFETY: all-zero is a valid `flock` value before named fields are set.
95        std::mem::zeroed::<libc::flock>()
96    };
97    lock.l_type = libc::F_WRLCK as libc::c_short;
98    lock.l_whence = libc::SEEK_SET as libc::c_short;
99    lock.l_start = 0;
100    lock.l_len = 0;
101    let result = unsafe {
102        // SAFETY: `file` owns a live descriptor and `lock` remains valid for
103        // the duration of this non-blocking `F_SETLK` call.
104        libc::fcntl(file.as_raw_fd(), libc::F_SETLK, &lock)
105    };
106    if result == -1 {
107        let error = std::io::Error::last_os_error();
108        if error
109            .raw_os_error()
110            .is_some_and(|errno| errno == libc::EAGAIN || errno == libc::EACCES)
111        {
112            Err(std::io::Error::from(std::io::ErrorKind::WouldBlock))
113        } else {
114            Err(error)
115        }
116    } else {
117        Ok(())
118    }
119}
120
121#[cfg(not(unix))]
122fn lock_exclusive(_: &File) -> std::io::Result<()> {
123    Err(std::io::Error::new(
124        std::io::ErrorKind::Unsupported,
125        "store locks require Unix fcntl record locks",
126    ))
127}
128
129impl Drop for StoreLock {
130    fn drop(&mut self) {
131        drop(self.file.take());
132        #[cfg(unix)]
133        held_stores().remove(&self.registry_key);
134    }
135}
136
137/// Failure to open or acquire the single-writer lock.
138#[derive(Debug)]
139pub enum StoreLockError {
140    /// A named filesystem or lock operation failed.
141    Io {
142        /// Lock-file path.
143        path: PathBuf,
144        /// Underlying platform error.
145        source: std::io::Error,
146    },
147}
148
149impl std::fmt::Display for StoreLockError {
150    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
151        match self {
152            Self::Io { path, source } => {
153                write!(formatter, "store lock {}: {source}", path.display())
154            }
155        }
156    }
157}
158
159impl std::error::Error for StoreLockError {
160    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
161        match self {
162            Self::Io { source, .. } => Some(source),
163        }
164    }
165}
zeppelin_embed/lifecycle/lock.rs

zeppelin_embed/lifecycle/
lock.rs
