hyperdir/
lock.rs

1//! S3-backed advisory file locks (the "global" lock mode).
2//!
3//! Two parts:
4//!
5//!   * A storage-agnostic byte-range lock engine ([`LockTable`]) with POSIX
6//!     semantics (read/write compatibility, overlap, replace/split of the
7//!     owner's own locks, group release, TTL expiry). Pure logic, unit-tested,
8//!     no I/O.
9//!   * A thin binding ([`HyperDir`] `fs_getlk` / `fs_setlk` / `fs_unlock_owner`
10//!     / `fs_lock_renew`) that persists one table per file/dir in a single
11//!     `<DIR|FILE/uuid>/_lock` object, updated with S3 conditional writes
12//!     (create-once `If-None-Match:*`, CAS `If-Match:<etag>`) under an OCC
13//!     retry loop — the same pattern as the compactor lease.
14//!
15//! The `_lock` object lives under the child's prefix, so it is reclaimed
16//! automatically when the file/dir prefix is deleted (no GC change needed).
17//!
18//! Crash release is by TTL: every held lock carries an absolute expiry; an
19//! expired entry is ignored and pruned on the next write. The holder must
20//! renew (`fs_lock_renew`) well within the TTL. Blocking acquisition
21//! (`F_SETLKW`) is not implemented here — the caller treats it as a single
22//! non-blocking attempt.
23//!
24//! `owner` is an opaque single-line token (no tab/newline). The FUSE layer
25//! folds its per-mount client id and the kernel `lock_owner` into it, so locks
26//! are distinguishable across mounts.
27
28use std::io::{Result, Error, ErrorKind};
29use aws_sdk_s3::Client;
30use aws_sdk_s3::primitives::SdkBody;
31use crate::s3::s3_error_kind;
32use uuid::Uuid;
33use crate::hyper::HyperDir;
34use crate::layout::HyperDirLayout;
35use crate::unix_now_ms;
36
37/// Object name (under each child's prefix) holding its lock table.
38const LOCK_OBJECT: &str = "_lock";
39/// Default lock lease TTL. A holder that crashes frees its locks within this.
40pub const DEFAULT_LOCK_TTL_MS: u64 = 60_000;
41/// Max optimistic-concurrency retries for a `_lock` read-modify-write.
42const LOCK_OCC_RETRIES: usize = 8;
43
44/// A lock's mode. `Unlock` is only ever passed to [`LockTable::set`]; it is
45/// never stored in an entry.
46#[derive(Clone, Copy, PartialEq, Eq, Debug)]
47pub enum LockKind {
48    Read,
49    Write,
50    Unlock,
51}
52
53impl LockKind {
54    fn as_char(self) -> char {
55        match self {
56            LockKind::Read => 'R',
57            LockKind::Write => 'W',
58            LockKind::Unlock => 'U',
59        }
60    }
61    fn from_char(c: char) -> Option<Self> {
62        match c {
63            'R' => Some(LockKind::Read),
64            'W' => Some(LockKind::Write),
65            _ => None,
66        }
67    }
68}
69
70/// An existing lock that conflicts with a requested one (returned by getlk /
71/// a failed setlk). Carries enough to fill a FUSE getlk reply.
72#[derive(Clone, Debug, PartialEq, Eq)]
73pub struct Conflict {
74    pub start: u64,
75    pub end: u64,
76    pub kind: LockKind,
77    pub pid: u32,
78}
79
80/// Outcome of a non-blocking setlk.
81#[derive(Debug, PartialEq, Eq)]
82pub enum SetLkOutcome {
83    Granted,
84    Conflict(Conflict),
85}
86
87#[derive(Clone, Debug)]
88struct Entry {
89    owner: String,
90    kind: LockKind, // Read | Write only
91    start: u64,
92    end: u64, // inclusive
93    pid: u32,
94    expire_ms: i64,
95}
96
97/// The set of locks held on one file/dir. Ranges are inclusive `[start, end]`.
98#[derive(Default, Debug)]
99pub struct LockTable {
100    entries: Vec<Entry>,
101}
102
103fn overlaps(a_start: u64, a_end: u64, b_start: u64, b_end: u64) -> bool {
104    a_start <= b_end && b_start <= a_end
105}
106
107impl LockTable {
108    fn purge_expired(&mut self, now_ms: i64) {
109        self.entries.retain(|e| e.expire_ms > now_ms);
110    }
111
112    /// Return a conflicting lock held by a *different* owner, or `None`.
113    /// `kind` is the requested mode (Read/Write). Expired entries are ignored.
114    pub fn test(&self, owner: &str, kind: LockKind, start: u64, end: u64, now_ms: i64) -> Option<Conflict> {
115        let want_write = kind == LockKind::Write;
116        for e in &self.entries {
117            if e.expire_ms <= now_ms || e.owner == owner {
118                continue;
119            }
120            if overlaps(start, end, e.start, e.end) && (want_write || e.kind == LockKind::Write) {
121                return Some(Conflict { start: e.start, end: e.end, kind: e.kind, pid: e.pid });
122            }
123        }
124        None
125    }
126
127    /// Drop/split the given owner's own locks within `[start, end]`.
128    fn remove_range(&mut self, owner: &str, start: u64, end: u64) {
129        let mut out = Vec::with_capacity(self.entries.len());
130        for e in std::mem::take(&mut self.entries) {
131            if e.owner != owner || !overlaps(start, end, e.start, e.end) {
132                out.push(e);
133                continue;
134            }
135            if e.start < start {
136                out.push(Entry { end: start - 1, ..e.clone() }); // start>=1 here
137            }
138            if e.end > end {
139                out.push(Entry { start: end + 1, ..e.clone() }); // end<u64::MAX here
140            }
141            // middle overlap is dropped
142        }
143        self.entries = out;
144    }
145
146    /// Apply a lock/unlock for `owner`. On `Unlock`, removes the owner's locks
147    /// in range. Otherwise, fails with the conflict if another owner holds an
148    /// incompatible overlapping lock; on success replaces the owner's own
149    /// locks in range with the new one.
150    #[allow(clippy::too_many_arguments)]
151    pub fn set(&mut self, owner: &str, kind: LockKind, start: u64, end: u64, pid: u32, expire_ms: i64, now_ms: i64)
152        -> std::result::Result<(), Conflict>
153    {
154        self.purge_expired(now_ms);
155        if kind == LockKind::Unlock {
156            self.remove_range(owner, start, end);
157            return Ok(());
158        }
159        if let Some(c) = self.test(owner, kind, start, end, now_ms) {
160            return Err(c);
161        }
162        self.remove_range(owner, start, end);
163        self.entries.push(Entry { owner: owner.to_string(), kind, start, end, pid, expire_ms });
164        Ok(())
165    }
166
167    /// Remove all of an owner's locks (FUSE release/flush). Returns true if
168    /// anything was removed.
169    pub fn release_owner(&mut self, owner: &str) -> bool {
170        let before = self.entries.len();
171        self.entries.retain(|e| e.owner != owner);
172        self.entries.len() != before
173    }
174
175    /// Refresh the expiry of all of an owner's locks. Returns true if the
176    /// owner held any.
177    pub fn renew(&mut self, owner: &str, expire_ms: i64) -> bool {
178        let mut any = false;
179        for e in &mut self.entries {
180            if e.owner == owner {
181                e.expire_ms = expire_ms;
182                any = true;
183            }
184        }
185        any
186    }
187
188    /// Encode as a line-based ASCII body: a `v1` header then one
189    /// tab-separated record per entry. `owner` must contain no tab/newline.
190    pub fn encode(&self) -> Vec<u8> {
191        let mut s = String::from("v1\n");
192        for e in &self.entries {
193            s.push_str(&format!(
194                "{}\t{}\t{}\t{}\t{}\t{}\n",
195                e.owner, e.kind.as_char(), e.start, e.end, e.pid, e.expire_ms
196            ));
197        }
198        s.into_bytes()
199    }
200
201    /// Decode [`encode`]'s format. Unparseable / unknown lines are skipped, so
202    /// a partial or future-versioned body degrades to "fewer locks" rather
203    /// than an error.
204    pub fn decode(buf: &[u8]) -> Self {
205        let mut t = LockTable::default();
206        let Ok(s) = std::str::from_utf8(buf) else { return t; };
207        for line in s.lines().skip(1) {
208            // skip "v1" header
209            let f: Vec<&str> = line.split('\t').collect();
210            if f.len() != 6 {
211                continue;
212            }
213            let (Some(kind), Ok(start), Ok(end), Ok(pid), Ok(expire_ms)) = (
214                f[1].chars().next().and_then(LockKind::from_char),
215                f[2].parse::<u64>(),
216                f[3].parse::<u64>(),
217                f[4].parse::<u32>(),
218                f[5].parse::<i64>(),
219            ) else {
220                continue;
221            };
222            t.entries.push(Entry { owner: f[0].to_string(), kind, start, end, pid, expire_ms });
223        }
224        t
225    }
226}
227
228/// S3 key of a child's lock-table object.
229fn lock_key(layout: &HyperDirLayout, uuid: &Uuid, is_dir: bool) -> String {
230    let base = if is_dir { layout.dir_key(uuid) } else { layout.file_key(uuid) };
231    format!("{}/{}", base, LOCK_OBJECT)
232}
233
234/// Load the lock table and its current ETag (`None` if the object doesn't
235/// exist yet).
236async fn load_table(client: &Client, bucket: &str, key: &str) -> Result<(LockTable, Option<String>)> {
237    match client.get_object().bucket(bucket).key(key).send().await {
238        Ok(res) => {
239            let etag = res.e_tag.clone().map(|t| t.replace('"', ""));
240            let bytes = res.body.collect().await
241                .map_err(|e| Error::other(format!("collect _lock s3://{}/{}: {}", bucket, key, e)))?
242                .into_bytes();
243            Ok((LockTable::decode(&bytes), etag))
244        },
245        Err(e) if e.as_service_error().is_some_and(|s| s.is_no_such_key()) => Ok((LockTable::default(), None)),
246        Err(e) => Err(Error::new(s3_error_kind(&e), format!("GetObject _lock s3://{}/{}: {}", bucket, key, e))),
247    }
248}
249
250enum PutOutcome {
251    Ok,
252    Retry, // CAS miss (412/409): table changed under us
253}
254
255/// Conditional PUT of the table body: create-once when `etag` is `None`,
256/// otherwise CAS on the ETag. A precondition failure is reported as `Retry`.
257async fn cas_put(client: &Client, bucket: &str, key: &str, body: Vec<u8>, etag: Option<&str>) -> Result<PutOutcome> {
258    let b = client.put_object().bucket(bucket).key(key).body(SdkBody::from(body).into());
259    let b = match etag {
260        Some(t) => b.if_match(t),
261        None => b.if_none_match('*'),
262    };
263    match b.send().await {
264        Ok(_) => Ok(PutOutcome::Ok),
265        Err(sdk_err) => {
266            if sdk_err.raw_response().map(|r| matches!(r.status().as_u16(), 412 | 409)).unwrap_or(false) {
267                Ok(PutOutcome::Retry)
268            } else {
269                Err(Error::new(s3_error_kind(&sdk_err), format!("PutObject _lock s3://{}/{}: {}", bucket, key, sdk_err)))
270            }
271        },
272    }
273}
274
275impl HyperDir<'_> {
276    /// Test for a conflicting lock without acquiring one (FUSE getlk).
277    #[allow(clippy::too_many_arguments)]
278    pub async fn fs_getlk(
279        client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool,
280        owner: &str, kind: LockKind, start: u64, end: u64,
281    ) -> Result<Option<Conflict>> {
282        let key = lock_key(layout, uuid, is_dir);
283        let (table, _) = load_table(client, bucket, &key).await?;
284        Ok(table.test(owner, kind, start, end, unix_now_ms()))
285    }
286
287    /// Acquire / change / release a byte-range lock (FUSE setlk, non-blocking).
288    /// Returns `Granted` or `Conflict`; transient CAS races are retried.
289    #[allow(clippy::too_many_arguments)]
290    pub async fn fs_setlk(
291        client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool,
292        owner: &str, kind: LockKind, start: u64, end: u64, pid: u32, ttl_ms: u64,
293    ) -> Result<SetLkOutcome> {
294        let key = lock_key(layout, uuid, is_dir);
295        for _ in 0..LOCK_OCC_RETRIES {
296            let (mut table, etag) = load_table(client, bucket, &key).await?;
297            // An unlock that finds no object has nothing to do.
298            if kind == LockKind::Unlock && etag.is_none() {
299                return Ok(SetLkOutcome::Granted);
300            }
301            let now = unix_now_ms();
302            let expire = now.saturating_add(ttl_ms as i64);
303            if let Err(c) = table.set(owner, kind, start, end, pid, expire, now) {
304                return Ok(SetLkOutcome::Conflict(c));
305            }
306            match cas_put(client, bucket, &key, table.encode(), etag.as_deref()).await? {
307                PutOutcome::Ok => return Ok(SetLkOutcome::Granted),
308                PutOutcome::Retry => continue,
309            }
310        }
311        Err(Error::new(ErrorKind::ResourceBusy, "fs_setlk: too many OCC conflicts"))
312    }
313
314    /// Release every lock held by `owner` on this file/dir (FUSE release/flush).
315    pub async fn fs_unlock_owner(
316        client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool, owner: &str,
317    ) -> Result<()> {
318        let key = lock_key(layout, uuid, is_dir);
319        for _ in 0..LOCK_OCC_RETRIES {
320            let (mut table, etag) = load_table(client, bucket, &key).await?;
321            let Some(etag) = etag else { return Ok(()); }; // no object => nothing held
322            if !table.release_owner(owner) {
323                return Ok(()); // we held nothing
324            }
325            match cas_put(client, bucket, &key, table.encode(), Some(&etag)).await? {
326                PutOutcome::Ok => return Ok(()),
327                PutOutcome::Retry => continue,
328            }
329        }
330        Err(Error::new(ErrorKind::ResourceBusy, "fs_unlock_owner: too many OCC conflicts"))
331    }
332
333    /// Refresh the TTL of every lock held by `owner` (maintenance renew).
334    /// A no-op if the owner holds nothing.
335    pub async fn fs_lock_renew(
336        client: &Client, layout: &HyperDirLayout, bucket: &str, uuid: &Uuid, is_dir: bool,
337        owner: &str, ttl_ms: u64,
338    ) -> Result<()> {
339        let key = lock_key(layout, uuid, is_dir);
340        for _ in 0..LOCK_OCC_RETRIES {
341            let (mut table, etag) = load_table(client, bucket, &key).await?;
342            let Some(etag) = etag else { return Ok(()); };
343            let expire = unix_now_ms().saturating_add(ttl_ms as i64);
344            if !table.renew(owner, expire) {
345                return Ok(());
346            }
347            match cas_put(client, bucket, &key, table.encode(), Some(&etag)).await? {
348                PutOutcome::Ok => return Ok(()),
349                PutOutcome::Retry => continue,
350            }
351        }
352        Err(Error::new(ErrorKind::ResourceBusy, "fs_lock_renew: too many OCC conflicts"))
353    }
354}
355
356#[cfg(test)]
357mod tests {
358    use super::*;
359
360    const FUTURE: i64 = i64::MAX;
361    const NOW: i64 = 1_000;
362
363    fn set(t: &mut LockTable, owner: &str, kind: LockKind, s: u64, e: u64) -> std::result::Result<(), Conflict> {
364        t.set(owner, kind, s, e, 0, FUTURE, NOW)
365    }
366
367    #[test]
368    fn read_read_compatible_write_excludes() {
369        let mut t = LockTable::default();
370        set(&mut t, "a", LockKind::Read, 0, 10).unwrap();
371        // another reader on overlapping range is fine
372        set(&mut t, "b", LockKind::Read, 5, 20).unwrap();
373        // a writer overlapping a reader conflicts
374        let c = set(&mut t, "c", LockKind::Write, 8, 9).unwrap_err();
375        assert_eq!(c.kind, LockKind::Read);
376        // a writer on a disjoint range is fine
377        set(&mut t, "c", LockKind::Write, 100, 200).unwrap();
378    }
379
380    #[test]
381    fn write_write_overlap_conflicts_disjoint_ok() {
382        let mut t = LockTable::default();
383        set(&mut t, "a", LockKind::Write, 0, 10).unwrap();
384        assert!(set(&mut t, "b", LockKind::Write, 10, 10).is_err()); // touch at 10
385        set(&mut t, "b", LockKind::Write, 11, 20).unwrap(); // adjacent, disjoint
386    }
387
388    #[test]
389    fn same_owner_replaces_and_upgrades() {
390        let mut t = LockTable::default();
391        set(&mut t, "a", LockKind::Read, 0, 10).unwrap();
392        // same owner upgrading its own range to write must not self-conflict
393        set(&mut t, "a", LockKind::Write, 0, 10).unwrap();
394        // and now another writer is excluded by the upgraded lock
395        assert!(set(&mut t, "b", LockKind::Write, 0, 0).is_err());
396    }
397
398    #[test]
399    fn unlock_splits_middle() {
400        let mut t = LockTable::default();
401        set(&mut t, "a", LockKind::Write, 0, 100).unwrap();
402        // punch out the middle
403        set(&mut t, "a", LockKind::Unlock, 40, 60).unwrap();
404        // left and right remain locked, middle is free
405        assert!(set(&mut t, "b", LockKind::Write, 50, 50).is_ok());
406        assert!(set(&mut t, "c", LockKind::Write, 0, 0).is_err());
407        assert!(set(&mut t, "d", LockKind::Write, 100, 100).is_err());
408    }
409
410    #[test]
411    fn release_owner_and_renew() {
412        let mut t = LockTable::default();
413        set(&mut t, "a", LockKind::Write, 0, 10).unwrap();
414        set(&mut t, "a", LockKind::Write, 20, 30).unwrap();
415        assert!(t.renew("a", FUTURE));
416        assert!(!t.renew("nobody", FUTURE));
417        assert!(t.release_owner("a"));
418        assert!(t.entries.is_empty());
419        assert!(set(&mut t, "b", LockKind::Write, 0, 30).is_ok());
420    }
421
422    #[test]
423    fn expired_locks_are_ignored() {
424        let mut t = LockTable::default();
425        // a's lock already expired (expire <= now)
426        t.set("a", LockKind::Write, 0, 10, 0, NOW - 1, NOW).unwrap();
427        // b can take it; the expired entry is pruned on this set
428        t.set("b", LockKind::Write, 0, 10, 0, FUTURE, NOW).unwrap();
429        assert!(t.test("c", LockKind::Read, 5, 5, NOW).is_some());
430    }
431
432    #[test]
433    fn encode_decode_roundtrip() {
434        let mut t = LockTable::default();
435        set(&mut t, "client-1:42", LockKind::Read, 0, 10).unwrap();
436        set(&mut t, "client-2:7", LockKind::Write, 100, u64::MAX).unwrap();
437        let back = LockTable::decode(&t.encode());
438        // semantics preserved: read region shared, write region exclusive
439        assert!(back.test("x", LockKind::Read, 0, 10, NOW).is_none());
440        assert!(back.test("x", LockKind::Write, 0, 0, NOW).is_some());
441        assert!(back.test("x", LockKind::Read, 200, 200, NOW).is_some());
442    }
443
444    #[test]
445    fn decode_tolerates_garbage() {
446        assert!(LockTable::decode(b"").entries.is_empty());
447        assert!(LockTable::decode(b"v1\ngarbage line\n\tbad\t\t\t\t").entries.is_empty());
448    }
449}
hyperdir/lock.rs

hyperdir/
lock.rs
