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

orbit_lock/
lib.rs
