SQLite engine: named locks — try_lock, SqliteLockHandle, duration guards (task sqlite-engine-locks)

This commit is contained in:
glm-5.3-flash committed 2026-10-08 13:37:22 +00:00
1 parent 513df0b311
commit 1edcb0e27d
5 files changed
+640 -13

No files matched your search

+427
View File
@@ -0,0 +1,427 @@
//! The named-lock mechanism's acceptance tests (`sqlite-engine-locks`):
//! exclusive grant over a temp-file store with the loser's `Ok(None)`
//! arm, release re-enabling acquisition, the duration guards (ADR-023
//! §2 — `ttl <= 0` rejected opaque on `try_lock` and `renew`, the
//! substrate's guard message preserved in the source chain on renew),
//! `renew`'s full-window-from-now semantics and lapsed-holder refusal,
//! the inherited same-owner re-acquire shape (granted, TTL not
//! refreshed — engine-sqlite.md's mapping row), `release`'s
//! owner-scoped delete + no-op false arm, the silent TTL lapse with
//! lapse-then-reacquire (ADR-008 §7's guarantee row), and the
//! engine-external lapse probe (a second `open` seeing the lapse
//! through the shared db file). ADR-019 §1–§2; ADR-023 §2.
use std::path::PathBuf;
use std::time::Duration;
use rusqlite::OptionalExtension;
use alkstore::{Error, Store};
use crate::store::open_store;
fn temp_dir(tag: &str) -> PathBuf {
let d = std::env::temp_dir().join(format!(
"alkstore-lock-{tag}-{}-{:?}",
std::process::id(),
std::thread::current().id()
));
let _ = std::fs::remove_dir_all(&d);
std::fs::create_dir_all(&d).unwrap();
d
}
fn temp_path(tag: &str) -> PathBuf {
temp_dir(tag).join("store.db")
}
fn now() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs() as i64
}
/// Wait until the lock row `name`'s expiry is in the past — the
/// bounded-wait posture (no tight-race assertions; second-precision
/// expiry means ~1 s of wall clock per step, well inside the bound).
async fn until_lapsed(path: &str, name: &str) {
let deadline = tokio::time::Instant::now() + Duration::from_secs(6);
loop {
let lapsed = {
let conn = rusqlite::Connection::open(path).unwrap();
let expires: Option<i64> = conn
.query_row(
"SELECT expires_at FROM __alkstore_locks WHERE name = ?1",
[name],
|r| r.get(0),
)
.optional()
.unwrap();
now() > expires.unwrap_or(i64::MAX)
};
if lapsed {
return;
}
assert!(
tokio::time::Instant::now() < deadline,
"lock {name}: never lapsed"
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
}
/// Acceptance: exclusive grant — the second acquirer gets `Ok(None)`
/// (no-work is a value, not an error), a different lock name is
/// independent, and release makes it acquirable again. The same-owner
/// re-acquire is the inherited shape: granted (`Some`), TTL **not**
/// refreshed (engine-sqlite.md's mapping row, inherited deliberately —
/// `renew` is the refresh path).
#[tokio::test(flavor = "multi_thread")]
async fn try_lock_grants_exclusively_and_release_reopens() {
let dir = temp_dir("exclusive");
let path = temp_path("exclusive");
let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
let a = store.try_lock("leader", "a", 300).await.unwrap();
assert!(a.is_some(), "first acquirer holds it");
let lock = a.unwrap();
assert_eq!(lock.name(), "leader");
let grant_expires: i64 = {
let conn = rusqlite::Connection::open(&path).unwrap();
conn.query_row(
"SELECT expires_at FROM __alkstore_locks WHERE name = 'leader'",
[],
|r| r.get(0),
)
.unwrap()
};
assert!(
store.try_lock("leader", "b", 300).await.unwrap().is_none(),
"someone holds it: Ok(None), never an error"
);
let reacquired = store
.try_lock("leader", "a", 300)
.await
.unwrap()
.expect("same-owner re-acquire is granted");
{
let conn = rusqlite::Connection::open(&path).unwrap();
let expires: i64 = conn
.query_row(
"SELECT expires_at FROM __alkstore_locks WHERE name = 'leader'",
[],
|r| r.get(0),
)
.unwrap();
assert!(
expires == grant_expires,
"re-acquire must not refresh the TTL (inherited deliberately): \
{expires} vs the grant's {grant_expires} — the row is the \
original insert, INSERT OR IGNORE keeps it"
);
}
assert!(
store.try_lock("leader", "b", 300).await.unwrap().is_none(),
"still exclusive after the no-refresh re-acquire"
);
assert!(
store
.try_lock("other-lock", "a", 300)
.await
.unwrap()
.is_some(),
"a different lock name is independent"
);
assert!(lock.release().await.unwrap(), "release reports it was held");
assert!(
store.try_lock("leader", "b", 300).await.unwrap().is_some(),
"release makes it acquirable again"
);
drop(reacquired);
store.close();
let _ = std::fs::remove_dir_all(dir);
}
/// Acceptance: the duration guards (ADR-023 §2) — `try_lock(ttl <= 0)`
/// and `renew(ttl <= 0)` are opaque `Err` (`Database`), never a
/// granted lock, never a panic. The engine-side guard's detail rides
/// the source chain (`#[source]`, ADR-008 §5); the substrate's own
/// guard (`ttl_s must be positive`) is preserved in the chain where
/// its call site fires — direct substrate callers meet it.
#[tokio::test(flavor = "multi_thread")]
async fn duration_guards_reject_non_positive_ttl_opaquely() {
let dir = temp_dir("duration");
let path = temp_path("duration");
let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
for ttl in [0, -1] {
let err = match store.try_lock("guarded", "a", ttl).await {
Ok(granted) => panic!(
"try_lock(ttl = {ttl}) must be a rejection, granted: {}",
granted.is_some()
),
Err(e) => e,
};
assert!(
matches!(err, Error::Database(_)),
"opaque Database shape, got: {err:?}"
);
assert!(
!matches!(err, Error::InvalidName { .. } | Error::ReservedName { .. }),
"the duration rule must not be misreported as a name error: {err:?}"
);
}
assert!(
store.try_lock("guarded", "a", 300).await.unwrap().is_some(),
"the guard fires before the substrate, so no lock row was created"
);
assert!(store.try_lock("guarded", "b", 300).await.unwrap().is_none());
let lock = store
.try_lock("renew-guard", "a", 300)
.await
.unwrap()
.unwrap();
for ttl in [0, -1] {
let err = match lock.renew(ttl).await {
Ok(landed) => panic!("renew({ttl}) must be a rejection, landed: {landed}"),
Err(e) => e,
};
assert!(matches!(err, Error::Database(_)), "opaque, got: {err:?}");
let detail = err.to_string();
assert!(
detail.contains("ttl must be a positive duration")
|| std::error::Error::source(&err)
.map(|s| s.to_string().contains("ttl must be a positive duration"))
.unwrap_or(false),
"the engine guard's detail rides the source chain, got: {detail}"
);
}
assert!(lock.renew(600).await.unwrap(), "a positive ttl renews");
// The substrate's guard message, preserved in the source chain
// where its call site fires: the direct substrate caller meets
// `ttl_s must be positive` verbatim.
let conn = rusqlite::Connection::open(&path).unwrap();
let err = crate::substrate::lock_renew(&conn, "renew-guard", "a", -1).unwrap_err();
assert_eq!(err.to_string(), "ttl_s must be positive");
store.close();
let _ = std::fs::remove_dir_all(dir);
}
/// Acceptance: `renew` is a new full TTL window from now (not
/// additive); a foreign owner's renew is `false` and leaves the row
/// untouched; a lapsed holder's renew is `false` (lost it — expired
/// and re-acquired elsewhere).
#[tokio::test(flavor = "multi_thread")]
async fn renew_is_full_window_and_refuses_foreign_or_lapsed_holders() {
let dir = temp_dir("renew");
let path = temp_path("renew");
let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
let lock = store.try_lock("leader", "a", 300).await.unwrap().unwrap();
// A foreign owner string's renew is false — the row predicate is
// (name, owner) — and the holder's renew still works after (the
// refusal never touched the row).
{
let conn = rusqlite::Connection::open(&path).unwrap();
assert_eq!(
crate::substrate::lock_renew(&conn, "leader", "nobody", 300).unwrap(),
0,
"foreign owner's renew is refused"
);
}
assert!(lock.renew(600).await.unwrap());
let before = now();
{
let conn = rusqlite::Connection::open(&path).unwrap();
let expires: i64 = conn
.query_row(
"SELECT expires_at FROM __alkstore_locks WHERE name = 'leader'",
[],
|r| r.get(0),
)
.unwrap();
assert!(
((before + 600 - 3)..=(before + 600 + 3)).contains(&expires),
"renew sets a full window from now (not additive): \
{expires} vs ~{before}+600"
);
}
// Lapse + re-acquire, the lapse-then-reacquire shape: stamp a
// short expiry directly (the honest cross-connection probe, at
// second precision), wait it out, re-acquire as a foreign owner,
// then the stale holder's renew is false (lost it).
{
let conn = rusqlite::Connection::open(&path).unwrap();
conn.execute(
"UPDATE __alkstore_locks SET expires_at = unixepoch() + 1
WHERE name = 'leader' AND owner = 'a'",
[],
)
.unwrap();
}
until_lapsed(path.to_str().unwrap(), "leader").await;
let _other = store
.try_lock("leader", "b", 1)
.await
.unwrap()
.expect("re-acquired after expiry");
until_lapsed(path.to_str().unwrap(), "leader").await;
assert!(!lock.renew(300).await.unwrap(), "lost it");
assert!(
!lock.release().await.unwrap(),
"the stale holder's release is the no-op arm after the foreign re-acquire"
);
store.close();
let _ = std::fs::remove_dir_all(dir);
}
/// Acceptance: `release` deletes only the owner's row (the
/// `(name, owner)` predicate — a foreign owner string deletes
/// nothing), and the no-op `false` arm: releasing with the row already
/// gone is `false`, not an error.
#[tokio::test(flavor = "multi_thread")]
async fn release_owner_scoped_and_missing_row_is_a_noop() {
let dir = temp_dir("release");
let path = temp_path("release");
let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
let lock = store.try_lock("leader", "a", 300).await.unwrap().unwrap();
// Substrate level: the delete predicate is (name, owner).
{
let conn = rusqlite::Connection::open(&path).unwrap();
assert_eq!(
crate::substrate::lock_acquire(&conn, "l2", "a", 300).unwrap(),
1
);
assert_eq!(
crate::substrate::lock_release(&conn, "l2", "b").unwrap(),
0,
"a foreign owner string must not delete the row"
);
assert!(
crate::substrate::lock_acquire(&conn, "l2", "b", 300).unwrap() == 0,
"the row survives a foreign owner's release attempt"
);
assert_eq!(
crate::substrate::lock_release(&conn, "l2", "a").unwrap(),
1,
"the owner's release removed the row"
);
assert!(
crate::substrate::lock_acquire(&conn, "l2", "b", 300).unwrap() == 1,
"acquirable again after the owner's release"
);
}
assert!(lock.release().await.unwrap(), "held: release deletes");
assert!(
store.try_lock("leader", "b", 300).await.unwrap().is_some(),
"acquirable right after the owner's release"
);
store.close();
let _ = std::fs::remove_dir_all(dir);
}
/// Acceptance: silent lapse (ADR-008 §7's guarantee row) — no
/// revocation event; the substrate's expired-row delete on the next
/// acquire is the lapse mechanism, pinned lapse-then-reacquire: after
/// expiry another owner acquires while the stale holder's ops
/// disown. Freshly after a lapse (row still present, not yet swept by
/// the foreign acquire) the stale holder's release deletes its own
/// stale row and reports `true` — the row-scoped delete is the
/// release mechanism; only a *removed* row releases as `false` (the
/// no-op arm).
#[tokio::test(flavor = "multi_thread")]
async fn ttl_lapses_silently_and_another_owner_acquires() {
let dir = temp_dir("lapse");
let path = temp_path("lapse");
let store = open_store(path.to_str().unwrap(), Default::default()).unwrap();
let lock = store.try_lock("leader", "a", 1).await.unwrap().unwrap();
until_lapsed(path.to_str().unwrap(), "leader").await;
// Lapse is silent: the stale holder's ops report through values,
// never an error or a revocation event.
assert!(
store.try_lock("leader", "b", 300).await.unwrap().is_some(),
"after expiry another owner acquires (the expired-row delete is the mechanism)"
);
assert!(
!lock.renew(300).await.unwrap(),
"the stale holder's renew is false: expired and re-acquired elsewhere"
);
assert!(
!lock.release().await.unwrap(),
"the stale holder's release is the no-op arm after the foreign re-acquire"
);
assert!(
store.try_lock("leader", "c", 300).await.unwrap().is_none(),
"b's fresh acquisition still excludes others"
);
// The no-op arm without a foreign re-acquire: lapse, the stale row
// still present, the holder releases — the owner-scoped delete
// lands (true), and the lock is immediately acquirable again.
let second = store.try_lock("silent", "a", 1).await.unwrap().unwrap();
until_lapsed(path.to_str().unwrap(), "silent").await;
assert!(second.release().await.unwrap(), "the stale row deletes");
assert!(
store.try_lock("silent", "b", 300).await.unwrap().is_some(),
"acquirable right after the stale holder's release"
);
store.close();
let _ = std::fs::remove_dir_all(dir);
}
/// Acceptance: the lapse is visible across engine instances — a second
/// `open` on the same db file (the engine-external probe) acquires
/// after the TTL lapses (the verification backlog's re-acquire-after-
/// expiry pin on SQLite; the substrate's expired-row delete is the
/// mechanism, exercised cross-connection).
#[tokio::test(flavor = "multi_thread")]
async fn re_acquired_after_expiry_from_a_second_engine_instance() {
let dir = temp_dir("reacquire");
let path = temp_path("reacquire");
let path_str = path.to_str().unwrap().to_string();
let store = open_store(&path_str, Default::default()).unwrap();
let lock = store.try_lock("leader", "a", 1).await.unwrap().unwrap();
let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
let (other, held) = loop {
let other = open_store(&path_str, Default::default()).unwrap();
match other.try_lock("leader", "b", 300).await.unwrap() {
Some(l) => break (other, l),
None => other.close(),
}
};
assert!(
tokio::time::Instant::now() < deadline,
"the lock never lapsed for the second instance"
);
assert_eq!(held.name(), "leader");
assert!(!lock.renew(300).await.unwrap(), "lapsed: renew refuses");
assert!(
!lock.release().await.unwrap(),
"lapsed + foreign re-acquire: the no-op arm"
);
store.close();
other.close();
let _ = std::fs::remove_dir_all(dir);
}