alkstore/alkstore-sqlite/src/lock.rs

//! The named-lock mechanism's SQLite arm (`sqlite-engine-locks`) — the
//! substrate's lock ops over the writer slot (locks are writer ops;
//! all three are single statements, auto-commit):
//!
//! - `try_lock(name, owner, ttl)` — the validated constructor:
//!   `name` shared-namespace (`validate_shared_name`), `owner`
//!   consumer-local non-empty only (ADR-019 §2 — `InvalidName`), the
//!   **duration guard (ADR-023 §2)** on `ttl <= 0` at the entry
//!   before the substrate. `Ok(None)` = someone holds it (no-work is
//!   a value, not an error); `Ok(Some(..))` = the boxed [`Lock`]
//!   handle.
//! - `Lock::renew(ttl)` — the same duration guard at the handle's
//!   entry (the substrate's `lock_renew` guard is preserved in the
//!   source chain where that call fires; the engine guards both sites
//!   uniformly so the rule is contract-side — the pg engine has no
//!   inherited guard at all and implements the same rule). Renew is a
//!   new **full** TTL window from now (`unixepoch() + ttl`); `false`
//!   = lost it (expired and re-acquired elsewhere — exclusion lapses
//!   silently at TTL expiry, no revocation event, ADR-008 §7).
//! - `Lock::release(self)` — consuming (`self: Box<Self>`, the RAII
//!   shape, ADR-019 §1); deletes only the owner's row; `false` =
//!   already expired/re-acquired (a no-op, not an error).

use std::sync::Arc;
use std::sync::atomic::AtomicBool;

use alkstore::{BoxedFuture, Lock, Result, validate_local_name, validate_shared_name};

use crate::seam::{closed_store_error, sqlite_error, with_writer};
use crate::substrate::{Writer, lock_acquire, lock_release, lock_renew};

/// 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 = concat!("__alkstore_", "scheduler");

/// The acquire path (see the module docs): validation, guard, then the
/// substrate's `lock_acquire` through the writer slot. `Ok(None)` is
/// the held-by-someone-else arm.
pub(crate) fn try_lock(
    name: &str,
    owner: &str,
    ttl: i64,
    writer: Arc<Writer>,
    closed: Arc<AtomicBool>,
) -> BoxedFuture<'static, Result<Option<Box<dyn Lock>>>> {
    if let Err(e) = validate_shared_name(name) {
        return Box::pin(async move { Err(e) });
    }
    if let Err(e) = validate_local_name(owner) {
        return Box::pin(async move { Err(e) });
    }
    if ttl <= 0 {
        return Box::pin(async move {
            Err(crate::seam::database_error(format!(
                "ttl must be a positive duration (seconds), got {ttl}"
            )))
        });
    }
    if closed.load(std::sync::atomic::Ordering::Acquire) {
        return Box::pin(async { Err(closed_store_error()) });
    }
    let name = name.to_string();
    let owner = owner.to_string();
    Box::pin(async move {
        let granted = with_writer(writer.clone(), {
            let name = name.clone();
            let owner = owner.clone();
            move |conn: &rusqlite::Connection| {
                Ok(lock_acquire(conn, &name, &owner, ttl).map_err(sqlite_error)? > 0)
            }
        })
        .await?;
        Ok(granted.then(|| {
            Box::new(SqliteLockHandle {
                name,
                owner,
                writer,
                closed,
            }) as Box<dyn Lock>
        }))
    })
}

/// The held named-lock handle (see the module docs). [`Lock::release`]
/// consumes it; `renew` is the repeatable `&self` case.
pub struct SqliteLockHandle {
    name: String,
    owner: String,
    writer: Arc<Writer>,
    closed: Arc<AtomicBool>,
}

impl std::fmt::Debug for SqliteLockHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SqliteLockHandle")
            .field("name", &self.name)
            .finish_non_exhaustive()
    }
}

impl SqliteLockHandle {
    pub(crate) fn new_internal(
        name: &str,
        owner: &str,
        writer: Arc<Writer>,
        closed: Arc<AtomicBool>,
    ) -> Self {
        Self {
            name: name.to_string(),
            owner: owner.to_string(),
            writer,
            closed,
        }
    }
}

impl Lock for SqliteLockHandle {
    fn name(&self) -> &str {
        &self.name
    }

    fn renew<'a>(&'a self, ttl: i64) -> BoxedFuture<'a, Result<bool>> {
        if ttl <= 0 {
            return Box::pin(async move {
                Err(crate::seam::database_error(format!(
                    "ttl must be a positive duration (seconds), got {ttl}"
                )))
            });
        }
        if self.closed.load(std::sync::atomic::Ordering::Acquire) {
            return Box::pin(async { Err(closed_store_error()) });
        }
        let name = self.name.clone();
        let owner = self.owner.clone();
        let writer = self.writer.clone();
        Box::pin(async move {
            with_writer(writer, move |conn| {
                Ok(lock_renew(conn, &name, &owner, ttl).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }

    fn release(self: Box<Self>) -> BoxedFuture<'static, Result<bool>> {
        let SqliteLockHandle {
            name,
            owner,
            writer,
            closed,
        } = *self;
        Box::pin(async move {
            if closed.load(std::sync::atomic::Ordering::Acquire) {
                return Err(closed_store_error());
            }
            with_writer(writer, move |conn| {
                Ok(lock_release(conn, &name, &owner).map_err(sqlite_error)? > 0)
            })
            .await
        })
    }
}