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

orbit_lock/
lib.rs
