//! The named-locks mechanism's mem arm — the in-memory TTL registry //! (engine-mem.md §Mapping the contract, locks row; ADR-008 §7's //! guarantee row). //! //! - Registry entries carry owner + deadline; expiry is a clock check //! at the try/renew boundaries (no ambient revoker exists). After //! TTL expiry exclusion lapses *silently* — no revocation event — //! and the name is re-acquirable; the holder may still release. //! - Contention on one store instance is serialized by the guarded //! section: first wins, the loser reads `None` cleanly (a value, //! never an error). A re-`try_lock` by the *same* owner while held //! grants (no second lease row exists to fight) without refreshing //! the deadline — the SQL engines' `INSERT OR IGNORE` + owner-match //! shape, mirrored. //! - **The duration guard at entry** (ADR-023 §2, durations kind): //! `ttl <= 0` rejects on `try_lock` *and* `renew` with the opaque //! `Database` error — never a silent immediately-expired lease. //! - Locks are not part of the tx seam (no `lock_tx` — contract //! verbatim): the overlay neither sees nor stages locks. The //! scheduler's leadership lock rides this registry through the //! engine-internal acquire/renew path (a reserved name consumer //! entry points can never collide with). use std::sync::Arc; use alkstore::{BoxedFuture, Error, Lock, Result, validate_local_name, validate_shared_name}; use crate::clock::Clock; use crate::state::Guard; use crate::state::{LockState, State}; /// The scheduler's leadership-lock name (ADR-009 §4 — the reserved- /// prefix derived name in the lock namespace; engine-internal: every /// consumer entry point rejects the same prefix, so a consumer can /// never collide with it). pub(crate) const SCHEDULER_LEADER_LOCK: &str = "__alkstore_scheduler"; fn duration_error(ttl: i64) -> Error { Error::database(std::io::Error::other(format!( "ttl must be a positive duration (seconds), got {ttl}" ))) } /// The registry acquire (the one owner of the row-level mechanical): /// purge the name's expired entry, grant iff the name is absent or /// already owned by this owner. `now` is the caller's clock read. pub(crate) fn lock_acquire(l: &mut LockState, name: &str, owner: &str, ttl: i64, now: i64) -> bool { if let Some(entry) = l.entries.get(name) && entry.expires_at <= now { l.entries.remove(name); } match l.entries.get(name) { Some(entry) => entry.owner == owner, None => { l.entries.insert( name.to_string(), crate::state::LockEntry { owner: owner.to_string(), expires_at: now.saturating_add(ttl), }, ); true } } } /// The registry renew: a new full TTL window from now. Refused /// (`false`) when the name is held by another owner, or past expiry — /// a renew can never revive a lapsed exclusion (the mem pin: renew /// past expiry is refused, the holder's window is the TTL's). pub(crate) fn lock_renew(l: &mut LockState, name: &str, owner: &str, ttl: i64, now: i64) -> bool { let Some(entry) = l.entries.get_mut(name) else { return false; }; if entry.owner != owner || entry.expires_at <= now { return false; } entry.expires_at = now.saturating_add(ttl); true } /// The registry release: delete the owner's row only. `false` = /// re-acquired elsewhere (a no-op, not an error). The holder may /// release past expiry — exclusion lapsed silently; releasing the /// lapsed row is still the holder's. pub(crate) fn lock_release(l: &mut LockState, name: &str, owner: &str) -> bool { match l.entries.get(name) { Some(entry) if entry.owner == owner => { l.entries.remove(name); true } _ => false, } } /// The consumer `try_lock` path (validation + duration guard + the /// guarded acquire; `Ok(None)` is the held-by-someone-else arm). pub(crate) fn try_lock( state: Arc, clock: Arc, name: &str, owner: &str, ttl: i64, ) -> Result>> { validate_shared_name(name)?; validate_local_name(owner)?; if ttl <= 0 { return Err(duration_error(ttl)); } let now = clock.unix_now(); let granted = { let mut l = state.locks.guard(); lock_acquire(&mut l, name, owner, ttl, now) }; if granted { Ok(Some(Box::new(MemLock { name: name.to_string(), owner: owner.to_string(), state, clock, }))) } else { Ok(None) } } /// The held named-lock handle (the repeatable `renew` case and the /// consuming `release`). pub struct MemLock { name: String, owner: String, state: Arc, clock: Arc, } impl std::fmt::Debug for MemLock { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("MemLock") .field("name", &self.name) .finish_non_exhaustive() } } impl MemLock { /// The engine-internal leadership lease (the scheduler's): the /// same registry machinery without consumer validation (the name /// is engine-derived and reserved). pub(crate) fn new_internal( name: &str, owner: &str, state: Arc, clock: Arc, ) -> Self { Self { name: name.to_string(), owner: owner.to_string(), state, clock, } } pub(crate) fn acquire_now(&self, ttl: i64, now: i64) -> bool { let mut l = self.state.locks.guard(); lock_acquire(&mut l, &self.name, &self.owner, ttl, now) } pub(crate) fn renew_at(&self, ttl: i64, now: i64) -> bool { let mut l = self.state.locks.guard(); lock_renew(&mut l, &self.name, &self.owner, ttl, now) } pub(crate) fn release_now(&self) -> bool { let mut l = self.state.locks.guard(); lock_release(&mut l, &self.name, &self.owner) } } impl Lock for MemLock { fn name(&self) -> &str { &self.name } fn renew<'a>(&'a self, ttl: i64) -> BoxedFuture<'a, Result> { if ttl <= 0 { return Box::pin(async move { Err(duration_error(ttl)) }); } let now = self.clock.unix_now(); let granted = self.renew_at(ttl, now); Box::pin(async move { Ok(granted) }) } fn release(self: Box) -> BoxedFuture<'static, Result> { Box::pin(async move { Ok(self.release_now()) }) } } #[cfg(test)] mod tests { use crate::store::MemStore; use alkstore::Store; #[tokio::test(flavor = "current_thread")] async fn contention_grants_once_and_the_loser_reads_none() { let store = MemStore::new(); store.clock.freeze(1_000); let first = store.try_lock("db", "w1", 60).await.unwrap(); assert!(first.is_some(), "the first acquire wins"); let loser = store.try_lock("db", "w2", 60).await.unwrap(); assert!( loser.is_none(), "the loser is a clean None (a value, never an error)" ); let same_owner = store.try_lock("db", "w1", 60).await.unwrap(); assert!( same_owner.is_some(), "the owner re-grants (no second lease row)" ); drop(first); drop(loser); if let Some(re) = same_owner { assert!(re.renew(60).await.unwrap(), "renew is an absolute reset"); } } #[tokio::test(flavor = "current_thread")] async fn duration_guards_reject_non_positive_ttls() { let store = MemStore::new(); for ttl in [0i64, -1] { match store.try_lock("db", "w", ttl).await { Err(alkstore::Error::Database(_)) => {} _other => panic!("try_lock guards a non-positive ttl"), } let granted = store.try_lock("db", "w", 60).await.unwrap().unwrap(); match granted.renew(ttl).await { Err(alkstore::Error::Database(_)) => {} _other => panic!("renew guards a non-positive ttl"), } } } #[tokio::test(flavor = "current_thread")] async fn expiry_lapses_silently_and_reacquires() { let store = MemStore::new(); store.clock.freeze(1_000); let holder = store.try_lock("db", "w1", 60).await.unwrap().unwrap(); assert!(store.try_lock("db", "w2", 60).await.unwrap().is_none()); store.clock.advance(59); assert!( store.try_lock("db", "w2", 60).await.unwrap().is_none(), "still within TTL" ); store.clock.advance(2); assert!( store.try_lock("db", "w2", 60).await.unwrap().is_some(), "exclusion lapsed silently — re-acquirable, no revocation event" ); assert!( !holder.renew(60).await.unwrap(), "the re-acquired lock is not the old holder's to renew" ); } #[tokio::test(flavor = "current_thread")] async fn renew_past_expiry_is_refused() { let store = MemStore::new(); store.clock.freeze(1_000); let holder = store.try_lock("db", "w1", 60).await.unwrap().unwrap(); store.clock.advance(61); assert!( !holder.renew(60).await.unwrap(), "a renew cannot revive a lapsed exclusion" ); let re = store.try_lock("db", "w2", 60).await.unwrap().unwrap(); assert!(re.release().await.unwrap(), "the new holder releases"); } #[tokio::test(flavor = "current_thread")] async fn release_shapes() { let store = MemStore::new(); store.clock.freeze(1_000); let holder = store.try_lock("db", "w1", 60).await.unwrap().unwrap(); assert!(holder.release().await.unwrap(), "a held lock releases true"); let w1 = store.try_lock("db", "w1", 60).await.unwrap().unwrap(); store.clock.advance(120); assert!( w1.release().await.unwrap(), "release past expiry stays the holder's — exclusion lapsed silently" ); let a = store.try_lock("db", "w1", 60).await.unwrap().unwrap(); store.clock.advance(120); store.try_lock("db", "w2", 60).await.unwrap().unwrap(); assert!( !a.release().await.unwrap(), "re-acquired elsewhere: the old holder's release is a no-op" ); } #[tokio::test(flavor = "current_thread")] async fn entry_validation() { let store = MemStore::new(); for bad in ["", " ", "__alkstore_scheduler"] { match store.try_lock(bad, "w", 60).await { Err(alkstore::Error::InvalidName { .. } | alkstore::Error::ReservedName { .. }) => { } _other => panic!("try_lock({bad:?}) rejects"), } } match store.try_lock("db", "", 60).await { Err(alkstore::Error::InvalidName { .. }) => {} _other => panic!("an empty owner rejects"), } } #[tokio::test(flavor = "current_thread")] async fn locks_are_not_tx_scoped() { let store = MemStore::new(); store.clock.freeze(1_000); assert!(store.try_lock("db", "w", 60).await.unwrap().is_some()); let tx = store.begin_tx().await.unwrap(); drop(tx); assert!( store.try_lock("db", "w2", 60).await.unwrap().is_none(), "rollback or not — the overlay never sees or stages locks" ); } }