orbit_lock/
lib.rs

1//! Fleet-shared keyed locks over Orbit shared memory.
2//!
3//! Current ownership lives in a fixed-capacity state table. A dedicated
4//! notified ring carries successful transitions for reactive consumers.
5
6mod error;
7mod fence;
8mod layout;
9mod protocol;
10mod state;
11
12use std::sync::{Arc, Mutex};
13use std::time::Duration;
14
15use bytes::Bytes;
16use orbit_core::fleet::FleetLaneCursor;
17use orbit_core::{Fleet, NetId64, RingLoss};
18
19pub use error::{Error, Result};
20pub use fence::{Fence, FenceToken};
21pub use layout::{
22    DefaultLockLayout, LOCK_EVENT_RING_CAPACITY, LOCK_EVENT_RING_KIND,
23    LOCK_EVENT_RING_PAYLOAD_CAPACITY, LOCK_EVENT_RING_SPEC, LOCK_STATE_KIND, LockLayout,
24};
25pub use state::{LOCK_STATE_CAPACITY, LOCK_STATE_PAYLOAD_MAX};
26
27#[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
28pub use orbit_core::RingEventFd;
29
30use layout::LockEventRecord;
31use state::LockStateStore;
32
33/// Typed namespace for lock keys.
34pub trait LockType {
35    const NAMESPACE: &'static str;
36}
37
38/// Resource identity used for contention.
39#[derive(Clone, Debug, PartialEq, Eq, Hash)]
40pub struct LockKey {
41    namespace: Bytes,
42    label: Bytes,
43}
44
45impl LockKey {
46    pub fn new<T: LockType>(label: impl Into<Bytes>) -> Self {
47        Self {
48            namespace: Bytes::from_static(T::NAMESPACE.as_bytes()),
49            label: label.into(),
50        }
51    }
52
53    /// Construct a dynamically namespaced key.
54    pub fn from_parts(namespace: impl Into<Bytes>, label: impl Into<Bytes>) -> Self {
55        Self {
56            namespace: namespace.into(),
57            label: label.into(),
58        }
59    }
60
61    pub fn namespace(&self) -> &[u8] {
62        &self.namespace
63    }
64
65    pub fn label(&self) -> &[u8] {
66        &self.label
67    }
68}
69
70/// Opaque caller-chosen identity for one lock tenure.
71#[derive(Clone, Debug, PartialEq, Eq, Hash)]
72pub struct LockOwner(Bytes);
73
74impl LockOwner {
75    pub fn new(value: impl Into<Bytes>) -> Self {
76        Self(value.into())
77    }
78
79    pub fn as_bytes(&self) -> &[u8] {
80        &self.0
81    }
82}
83
84impl From<Bytes> for LockOwner {
85    fn from(value: Bytes) -> Self {
86        Self(value)
87    }
88}
89
90impl From<String> for LockOwner {
91    fn from(value: String) -> Self {
92        Self(Bytes::from(value))
93    }
94}
95
96impl From<&str> for LockOwner {
97    fn from(value: &str) -> Self {
98        Self(Bytes::copy_from_slice(value.as_bytes()))
99    }
100}
101
102/// One current lock tenure returned by the authoritative table.
103#[derive(Clone, Debug, PartialEq, Eq)]
104pub struct LockLease {
105    pub lock_id: NetId64,
106    pub key: LockKey,
107    pub owner: LockOwner,
108    /// Host-monotonic milliseconds at which this tenure began.
109    pub acquired_at_ms: u64,
110    /// Host-monotonic expiry deadline.
111    pub expires_at_ms: u64,
112    pub state_revision: u64,
113}
114
115impl LockLease {
116    /// Strictly increasing token for successive lock tenures.
117    pub fn fencing_token(&self) -> FenceToken {
118        FenceToken::new(self.lock_id.counter())
119    }
120
121    pub fn is_expired_at(&self, now_ms: u64) -> bool {
122        self.expires_at_ms <= now_ms
123    }
124}
125
126/// Result of one atomic acquisition attempt.
127#[derive(Clone, Debug, PartialEq, Eq)]
128pub enum LockAcquire {
129    Acquired(LockLease),
130    Occupied(LockLease),
131}
132
133impl LockAcquire {
134    pub fn acquired(self) -> Option<LockLease> {
135        match self {
136            Self::Acquired(lease) => Some(lease),
137            Self::Occupied(_) => None,
138        }
139    }
140
141    pub fn holder(&self) -> &LockLease {
142        match self {
143            Self::Acquired(lease) | Self::Occupied(lease) => lease,
144        }
145    }
146}
147
148/// One successful current-state change.
149#[derive(Clone, Debug, PartialEq, Eq)]
150pub enum LockTransition {
151    Acquired(LockLease),
152    Renewed(LockLease),
153    Released {
154        lock_id: NetId64,
155        key: LockKey,
156        owner: LockOwner,
157        state_revision: u64,
158    },
159}
160
161impl LockTransition {
162    pub fn lock_id(&self) -> NetId64 {
163        match self {
164            Self::Acquired(lease) | Self::Renewed(lease) => lease.lock_id,
165            Self::Released { lock_id, .. } => *lock_id,
166        }
167    }
168
169    pub fn key(&self) -> &LockKey {
170        match self {
171            Self::Acquired(lease) | Self::Renewed(lease) => &lease.key,
172            Self::Released { key, .. } => key,
173        }
174    }
175
176    pub fn owner(&self) -> &LockOwner {
177        match self {
178            Self::Acquired(lease) | Self::Renewed(lease) => &lease.owner,
179            Self::Released { owner, .. } => owner,
180        }
181    }
182
183    pub fn state_revision(&self) -> u64 {
184        match self {
185            Self::Acquired(lease) | Self::Renewed(lease) => lease.state_revision,
186            Self::Released { state_revision, .. } => *state_revision,
187        }
188    }
189}
190
191/// Transition decoded from the notified ring.
192#[derive(Clone, Debug, PartialEq, Eq)]
193pub struct LockEvent {
194    pub event_id: NetId64,
195    pub transition: LockTransition,
196}
197
198#[derive(Clone, Debug, PartialEq, Eq)]
199pub struct LockCursor {
200    inner: FleetLaneCursor,
201}
202
203/// Result of advancing one lock-event cursor.
204#[derive(Clone, Debug, Default, PartialEq, Eq)]
205pub struct LockPoll {
206    pub events: Vec<LockEvent>,
207    pub loss: RingLoss,
208    pub malformed: u64,
209}
210
211impl LockPoll {
212    pub fn is_empty(&self) -> bool {
213        self.events.is_empty() && self.loss.is_empty() && self.malformed == 0
214    }
215}
216
217/// Fleet-shared lock handle. Clones share one event cursor.
218#[derive(Clone)]
219pub struct Lock<L: LockLayout = DefaultLockLayout> {
220    fleet: Arc<Fleet>,
221    state: LockStateStore<L>,
222    cursor: Arc<Mutex<LockCursor>>,
223}
224
225impl Lock<DefaultLockLayout> {
226    /// Subscribe after all currently committed transition events.
227    pub fn new(fleet: Arc<Fleet>) -> Result<Self> {
228        Self::open(fleet, false)
229    }
230
231    /// Replay retained transitions. Intended for tests and diagnostics.
232    pub fn replay_retained(fleet: Arc<Fleet>) -> Result<Self> {
233        Self::open(fleet, true)
234    }
235}
236
237impl<L: LockLayout> Lock<L> {
238    /// Subscribe with a custom storage layout.
239    pub fn with_layout(fleet: Arc<Fleet>) -> Result<Self> {
240        Self::open(fleet, false)
241    }
242
243    /// Replay retained transitions with a custom storage layout.
244    /// Intended for tests and diagnostics.
245    pub fn replay_layout(fleet: Arc<Fleet>) -> Result<Self> {
246        Self::open(fleet, true)
247    }
248
249    fn open(fleet: Arc<Fleet>, replay_retained: bool) -> Result<Self> {
250        layout::validate::<L>()?;
251        let state = LockStateStore::new(&fleet);
252        let cursor = LockCursor {
253            inner: if replay_retained {
254                fleet.lane_cursor_from_start::<LockEventRecord<L>>()
255            } else {
256                fleet.lane_cursor_at_head::<LockEventRecord<L>>()
257            },
258        };
259        Ok(Self {
260            fleet,
261            state,
262            cursor: Arc::new(Mutex::new(cursor)),
263        })
264    }
265
266    pub fn try_acquire(
267        &self,
268        key: &LockKey,
269        owner: &LockOwner,
270        ttl: Duration,
271    ) -> Result<LockAcquire> {
272        let now_ms = monotonic_ms()?;
273        let expires_at_ms = deadline(now_ms, ttl)?;
274        self.state
275            .acquire(key, owner, now_ms, expires_at_ms, |transition| {
276                self.publish(transition).map(|_| ())
277            })
278    }
279
280    /// Renew only if this exact tenure is still current.
281    pub fn renew(&self, lease: &LockLease, ttl: Duration) -> Result<Option<LockLease>> {
282        let now_ms = monotonic_ms()?;
283        let expires_at_ms = deadline(now_ms, ttl)?;
284        self.state
285            .renew_id(lease, now_ms, expires_at_ms, |transition| {
286                self.publish(transition).map(|_| ())
287            })
288    }
289
290    /// Renew by a caller-retained owner identity.
291    pub fn renew_owned(
292        &self,
293        key: &LockKey,
294        owner: &LockOwner,
295        ttl: Duration,
296    ) -> Result<Option<LockLease>> {
297        let now_ms = monotonic_ms()?;
298        let expires_at_ms = deadline(now_ms, ttl)?;
299        self.state
300            .renew_owner(key, owner, now_ms, expires_at_ms, |transition| {
301                self.publish(transition).map(|_| ())
302            })
303    }
304
305    /// Release only if this exact tenure is still current.
306    pub fn release(&self, lease: &LockLease) -> Result<bool> {
307        let now_ms = monotonic_ms()?;
308        self.state.release_id(lease, now_ms, |transition| {
309            self.publish(transition).map(|_| ())
310        })
311    }
312
313    /// Release only when the current owner bytes match.
314    pub fn release_owned(&self, key: &LockKey, owner: &LockOwner) -> Result<bool> {
315        let now_ms = monotonic_ms()?;
316        self.state.release_owner(key, owner, now_ms, |transition| {
317            self.publish(transition).map(|_| ())
318        })
319    }
320
321    /// Release the current tenure without checking its owner.
322    ///
323    /// This is the explicit administrative escape hatch required by lock
324    /// providers such as Laravel's `forceRelease`. Normal callers should use
325    /// [`Self::release`] or [`Self::release_owned`].
326    pub fn force_release(&self, key: &LockKey) -> Result<bool> {
327        let now_ms = monotonic_ms()?;
328        self.state.force_release(key, now_ms, |transition| {
329            self.publish(transition).map(|_| ())
330        })
331    }
332
333    /// Read authoritative current state for one key.
334    pub fn current(&self, key: &LockKey) -> Result<Option<LockLease>> {
335        self.state.current(key, monotonic_ms()?)
336    }
337
338    /// Restore the current tenure only when its owner bytes match.
339    pub fn restore(&self, key: &LockKey, owner: &LockOwner) -> Result<Option<LockLease>> {
340        Ok(self
341            .current(key)?
342            .filter(|lease| lease.owner.as_bytes() == owner.as_bytes()))
343    }
344
345    /// Drain every currently visible transition from all writer lanes.
346    pub fn poll(&self) -> LockPoll {
347        let raw = {
348            let mut cursor = self
349                .cursor
350                .lock()
351                .unwrap_or_else(|error| error.into_inner());
352            self.fleet
353                .poll_lanes::<LockEventRecord<L>>(&mut cursor.inner)
354        };
355        let mut events = Vec::with_capacity(raw.frames.len());
356        let mut malformed = 0;
357        for frame in raw.frames {
358            match protocol::decode(&frame) {
359                Some(event) => events.push(event),
360                None => malformed += 1,
361            }
362        }
363        events.sort_by_key(|event| event.transition.state_revision());
364        LockPoll {
365            events,
366            loss: raw.loss,
367            malformed,
368        }
369    }
370
371    /// Clear lock state and transition history during quiescent owner boot.
372    pub fn reset_transport(&self) -> Result<()> {
373        let mut cursor = self
374            .cursor
375            .lock()
376            .unwrap_or_else(|error| error.into_inner());
377        self.state.reset()?;
378        self.fleet.reset_ring::<LockEventRecord<L>>()?;
379        cursor.inner = self.fleet.lane_cursor_at_head::<LockEventRecord<L>>();
380        Ok(())
381    }
382
383    /// Remove the state and event SHM objects.
384    #[cfg(unix)]
385    pub fn unlink(&self) -> Result<()> {
386        let ring = self.fleet.shm_ring::<LockEventRecord<L>>()?;
387        let ring_result = ring.unlink();
388        let state_result = self.state.unlink();
389        ring_result?;
390        state_result
391    }
392
393    /// Create this process' readiness fd for lock transitions.
394    #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
395    pub fn event_fd(&self) -> Result<RingEventFd> {
396        self.fleet
397            .ring_event_fd::<LockEventRecord<L>>()
398            .map_err(Error::Io)
399    }
400
401    fn publish(&self, transition: &LockTransition) -> Result<LockEvent> {
402        let (kind, payload) = protocol::encode::<L>(transition)?;
403        let event_id = {
404            #[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "macos"))]
405            {
406                self.fleet.publish_notified::<LockEventRecord<L>>(
407                    kind,
408                    transition.state_revision(),
409                    payload,
410                )?
411            }
412            #[cfg(not(any(target_os = "linux", target_os = "freebsd", target_os = "macos")))]
413            {
414                self.fleet
415                    .publish::<LockEventRecord<L>>(kind, transition.state_revision(), payload)
416            }
417        };
418        Ok(LockEvent {
419            event_id,
420            transition: transition.clone(),
421        })
422    }
423}
424
425fn deadline(now_ms: u64, ttl: Duration) -> Result<u64> {
426    if ttl.is_zero() {
427        return Err(Error::TtlZero);
428    }
429    let ttl_ms = u64::try_from(ttl.as_millis().max(1)).map_err(|_| Error::TtlOverflow)?;
430    now_ms.checked_add(ttl_ms).ok_or(Error::TtlOverflow)
431}
432
433#[cfg(unix)]
434fn monotonic_ms() -> Result<u64> {
435    let mut value = std::mem::MaybeUninit::<libc::timespec>::uninit();
436    let result = unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, value.as_mut_ptr()) };
437    if result != 0 {
438        return Err(Error::Io(std::io::Error::last_os_error()));
439    }
440    let value = unsafe { value.assume_init() };
441    let seconds = u64::try_from(value.tv_sec).map_err(|_| Error::TtlOverflow)?;
442    let nanos = u64::try_from(value.tv_nsec).map_err(|_| Error::TtlOverflow)?;
443    Ok(seconds
444        .saturating_mul(1_000)
445        .saturating_add(nanos / 1_000_000))
446}
447
448#[cfg(not(unix))]
449fn monotonic_ms() -> Result<u64> {
450    use std::sync::LazyLock;
451    use std::time::Instant;
452
453    static START: LazyLock<Instant> = LazyLock::new(Instant::now);
454    Ok(START.elapsed().as_millis().min(u128::from(u64::MAX)) as u64)
455}
orbit_lock/lib.rs

orbit_lock/
lib.rs
